Skip to content
43 changes: 37 additions & 6 deletions scripts/extract_api_model.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
import json
import re
import sys
import textwrap

PACKAGE = "bedrock_agentcore"

Expand All @@ -43,15 +44,33 @@
("knowledge-base", "Knowledge Base", "bedrock_agentcore.knowledge_base"),
]

_SECTION_RE = re.compile(r"^\s*(Args|Arguments|Returns|Raises|Example|Examples):\s*$")
_ARG_RE = re.compile(r"^\s+(\w+)\s*(?:\(([^)]+)\))?:\s*(.*)$")
_SECTION_RE = re.compile(r"^\s*(Args|Arguments|Returns|Raises|Example|Examples)(?:\s+\([^)]+\))?:\s*$")
_ARG_RE = re.compile(r"^\s+(\*{0,2}\w+)\s*(?:\(([^)]+)\))?:\s*(.*)$")

# The SDK mixes Google-style ("Args:") and reST-style (":param x:") docstrings,
# so we also recognize the reST field forms and pull them out of the prose.
_REST_PARAM_RE = re.compile(r"^\s*:param\s+(\w+):\s*(.*)$")
_REST_RETURNS_RE = re.compile(r"^\s*:returns?:\s*(.*)$")
_REST_RAISES_RE = re.compile(r"^\s*:raises?\s+([\w.]+):\s*(.*)$")

SUMMARY_OVERRIDES = {
"Actor": (
"Provides a handle for an actor within a session, delegating operations to the associated MemorySessionManager."
),
"ActorProfile": "Describes the simulated actor's identity and objective.",
"AgentCoreRuntimeClient": "Generates WebSocket authentication for Amazon Bedrock AgentCore runtime.",
"BatchEvaluationSummary": "Provides aggregated results from a completed batch evaluation.",
"CodeInterpreter": "Provides a client for the Amazon Bedrock AgentCore Code Interpreter sandbox service.",
"ConfigBundleRef": "References a configuration bundle version parsed from OTEL baggage.",
"MemorySession": "Represents a single Amazon Bedrock AgentCore MemorySession resource.",
"RuntimeClient": "Generates WebSocket authentication for Amazon Bedrock AgentCore runtime.",
"delete_all_long_term_memories_in_namespace": ("Deletes all long-term memory records in the specified namespace."),
}

DESCRIPTION_OVERRIDES = {
"delete_all_long_term_memories_in_namespace": "",
}


def extract_rest_fields(lines, result):
"""Pull reST field lines (:param:/:returns:/:raises:) out of `lines`.
Expand Down Expand Up @@ -100,7 +119,8 @@ def parse_google_docstring(doc):
result["summary"] = " ".join(summary)

section = "description"
desc, example_buf = [], []
desc, example_bufs = [], []
example_buf = None
while i < len(lines):
line = lines[i]
m = _SECTION_RE.match(line)
Expand All @@ -114,6 +134,9 @@ def parse_google_docstring(doc):
"example": "example",
"examples": "example",
}[name]
if section == "example":
example_buf = []
example_bufs.append(example_buf)
i += 1
continue
if section == "description":
Expand All @@ -125,7 +148,7 @@ def parse_google_docstring(doc):
{
"name": am.group(1),
"type": (am.group(2) or "").strip() or None,
"required": "optional" not in (am.group(2) or "").lower(),
"required": not am.group(1).startswith("*") and "optional" not in (am.group(2) or "").lower(),
"description": am.group(3).strip(),
}
)
Expand All @@ -141,6 +164,8 @@ def parse_google_docstring(doc):
am = _ARG_RE.match(line)
if am:
result["raises"].append({"type": am.group(1), "description": am.group(3).strip()})
elif result["raises"] and line.strip():
result["raises"][-1]["description"] += " " + line.strip()
elif section == "example":
example_buf.append(line)
i += 1 # always advance — non-header branches above don't, else infinite loop
Expand All @@ -149,8 +174,8 @@ def parse_google_docstring(doc):
# instead of, or mixed with, Google sections. Pull those out of the prose.
desc = extract_rest_fields(desc, result)
result["description"] = "\n".join(desc).strip()
if example_buf:
code = "\n".join(example_buf).strip()
for example_buf in example_bufs:
code = textwrap.dedent("\n".join(example_buf)).strip()
# strip a leading ```python fence if the docstring used one
code = re.sub(r"^```\w*\n?|\n?```$", "", code).strip()
if code:
Expand All @@ -177,6 +202,12 @@ def _own_docstring(obj):
def entry_from_object(name, obj):
"""Build a doc-model entry for a class or function."""
doc = parse_google_docstring(_own_docstring(obj))
if name in SUMMARY_OVERRIDES:
doc["summary"] = SUMMARY_OVERRIDES[name]
if name in DESCRIPTION_OVERRIDES:
doc["description"] = DESCRIPTION_OVERRIDES[name]
if name == "__init__" and doc["summary"] == "Represents an actor within a session.":
doc["summary"] = "Initializes an Actor instance for the specified session."
try:
signature = inspect.signature(obj)
# Drop the implicit `self`/`cls` receiver from method signatures.
Expand Down
102 changes: 99 additions & 3 deletions scripts/render_adoc.py
Original file line number Diff line number Diff line change
Expand Up @@ -55,12 +55,108 @@
SCHEMA_VERSION = 1


def normalize_style(text):
"""Apply style-safe substitutions to generated prose."""
if not text:
return ""
text = re.sub(r"\be\.g\.(?:,)?", "for example,", text, flags=re.IGNORECASE)
text = re.sub(
r"\bAWS (?:Bedrock(?: AgentCore)? )?Code\s*Interpreter\b",
"Amazon Bedrock AgentCore Code Interpreter",
text,
flags=re.IGNORECASE,
)
text = text.replace(
"Bedrock AgentCore Policy Engine client.",
"Policy Engine client for Amazon Bedrock AgentCore.",
)
text = text.replace(
"Client for Bedrock AgentCore Policy Engine operations.",
"Provides a client for Policy in AgentCore.",
)
text = re.sub(r"\bAWS Bedrock AgentCore\b", "Amazon Bedrock AgentCore", text)
text = re.sub(r"\bAWS Bedrock\b", "Amazon Bedrock", text)
text = re.sub(
r"(?<!Amazon )(?<!AWS )\bBedrock AgentCore\b",
"Amazon Bedrock AgentCore",
text,
)
text = re.sub(
r"(?<!Amazon Bedrock )\bAgentCore Code Interpreter\b",
"Amazon Bedrock AgentCore Code Interpreter",
text,
)
text = re.sub(
r"(?<!Amazon Bedrock )\bAgentCore runtime\b",
"Amazon Bedrock AgentCore runtime",
text,
)
text = re.sub(
r"(?<!Amazon Bedrock )\bAgentCore Identity\b",
"Amazon Bedrock AgentCore Identity",
text,
)
text = text.replace(", allowing applications to", " so applications can")
text = re.sub(r"\bAWS region\b", "AWS Region", text, flags=re.IGNORECASE)
text = re.sub(r"\bAgentCore Memory\b", "AgentCore memory", text)
text = re.sub(r"\bAgentCore Runtime\b", "AgentCore runtime", text)
text = text.replace(
"BedrockAgentCore Runtime Package.",
"Amazon Bedrock AgentCore runtime package.",
)
text = re.sub(r"\bAgentCore SDK\b", "AgentCore Python SDK", text)
text = text.replace(
"This feature is in preview and may change in future releases.",
"This feature is in preview and might change in future releases.",
)
text = text.replace("validation will ensure", "validation ensures")
text = text.replace(
"Delete all long-term memory records within a specific namespace.",
"Deletes all long-term memory records in the specified namespace.",
)
text = text.replace(
"This class provides convenient delegation to MemorySessionManager operations.",
"Use this class to delegate operations to MemorySessionManager.",
)
return re.sub(r"\bAWS\b", "{aws}", text)


def normalize_param_description(text):
"""Normalize recurring parameter-description style issues."""
text = normalize_style(clean_rst(text)).strip()
optional_replacement = "Optional" if _starts_with_plural_noun(text) else "An optional"
substitutions = (
(r"^Optional\b", optional_replacement),
(r"^(?:\{aws\}|AWS)\s+region\b", "The {aws} Region"),
(r"^id of\b", "The ID of"),
(r"^Behaviour\b", "The behavior"),
(r"^Behavior\b", "The behavior"),
(r"^Memory resource ID\b", "The memory resource ID"),
(r"^Strategy name\b", "The name of the memory strategy"),
(r"^Strategy ID\b", "The ID of the memory strategy"),
)
for pattern, replacement in substitutions:
text = re.sub(pattern, replacement, text, count=1, flags=re.IGNORECASE)
return text


def _starts_with_plural_noun(text):
"""Return whether an Optional description starts with a likely plural noun."""
match = re.match(r"^Optional\s+([A-Za-z]+)\b", text, flags=re.IGNORECASE)
if not match:
return False
noun = match.group(1)
return noun.islower() and noun.endswith("s") and not noun.endswith(("is", "ss", "us"))


def esc(text):
"""Escape AsciiDoc-significant characters in inline text."""
if not text:
return ""
# Guard the couple of chars that start AsciiDoc markup in running prose.
return text.replace("|", "\\|").replace("{", "\\{")
marker = "\0AWS_ENTITY\0"
text = normalize_style(text).replace("{aws}", marker)
return text.replace("|", "\\|").replace("{", "\\{").replace(marker, "{aws}")


# Match markdown code fences that may be indented (reST/Google docstrings often
Expand Down Expand Up @@ -94,7 +190,7 @@ def _admonition_repl(m):
# Convert roles: :class:`Foo` -> `Foo`
text = _RST_ROLE_RE.sub(r"`\1`", text)

return text
return normalize_style(text)


_ADOC_ADMONITION_RE = re.compile(
Expand Down Expand Up @@ -171,7 +267,7 @@ def render_params(params, out):
req = "" if p.get("required") else " _(optional)_"
typ = f"`{p['type']}`" if p.get("type") else ""
out.append(f"`{p['name']}`{req} {typ}::")
out.append(esc(clean_rst(p.get("description", ""))) or "_No description._")
out.append(esc(normalize_param_description(p.get("description", ""))) or "_No description._")
out.append("")


Expand Down
87 changes: 87 additions & 0 deletions tests/unit/scripts/test_extract_api_model.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,87 @@
"""Tests for the Python API doc-model extractor."""

import importlib.util
from pathlib import Path

_EXTRACT_PATH = Path(__file__).resolve().parents[3] / "scripts" / "extract_api_model.py"
_spec = importlib.util.spec_from_file_location("extract_api_model", _EXTRACT_PATH)
extract_api_model = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(extract_api_model)


def test_multiline_fields_and_named_examples_are_preserved():
doc = """Run an evaluation.

Args:
wait_config: Optional WaitConfig for polling behavior.
*args: Positional arguments forwarded to the API.
**kwargs: Arguments forwarded to the API.

Returns:
A list of spans.

Raises:
ValueError: If the dataset is empty or all scenarios fail during
execution.

Example (Runtime agent):
>>> run("runtime")
>>> run("runtime-again")

Example (Custom agent):
>>> run("custom")
"""

parsed = extract_api_model.parse_google_docstring(doc)

assert [param["name"] for param in parsed["params"]] == ["wait_config", "*args", "**kwargs"]
assert [param["required"] for param in parsed["params"]] == [True, False, False]
assert parsed["returns"]["description"] == "A list of spans."
assert parsed["raises"][0]["description"] == ("If the dataset is empty or all scenarios fail during execution.")
assert [example["code"] for example in parsed["examples"]] == [
'>>> run("runtime")\n>>> run("runtime-again")',
'>>> run("custom")',
]


def test_public_class_summaries_use_action_verbs():
class Actor:
"""Represents an actor within a session."""

Actor.__module__ = "bedrock_agentcore.memory"

entry = extract_api_model.entry_from_object("Actor", Actor)

assert entry["summary"].startswith("Provides a handle")


def test_internal_batch_size_is_removed_from_public_method_description():
def delete_all_long_term_memories_in_namespace():
"""Delete all records.

This method processes records in chunks of 100.
"""

entry = extract_api_model.entry_from_object(
"delete_all_long_term_memories_in_namespace",
delete_all_long_term_memories_in_namespace,
)

assert entry["summary"] == "Deletes all long-term memory records in the specified namespace."
assert entry["description"] == ""


def test_public_service_classes_use_full_service_names():
class MemorySession:
"""Represents a single, AgentCore MemorySession resource."""

class CodeInterpreter:
"""Client for interacting with the AgentCore Code Interpreter sandbox service."""

memory_entry = extract_api_model.entry_from_object("MemorySession", MemorySession)
interpreter_entry = extract_api_model.entry_from_object("CodeInterpreter", CodeInterpreter)

assert memory_entry["summary"] == "Represents a single Amazon Bedrock AgentCore MemorySession resource."
assert interpreter_entry["summary"] == (
"Provides a client for the Amazon Bedrock AgentCore Code Interpreter sandbox service."
)
Loading
Loading