aaa.mcp

MCPClient

MCPClient(
    session: Any | None = None,
    *,
    name: str = "mcp",
    _connect: Callable[[AsyncExitStack], Any] | None = None,
)

One connection to one MCP server. Use as an async context manager.

stdio classmethod

stdio(
    command: str,
    args: Sequence[str] = (),
    *,
    env: Mapping[str, str] | None = None,
    cwd: str | None = None,
    name: str | None = None,
) -> MCPClient

Launch command as a subprocess and speak MCP over its stdin/stdout. The child gets only a small safe set of environment variables plus env: put secrets in env explicitly.

http classmethod

http(
    url: str,
    *,
    headers: Mapping[str, str] | None = None,
    name: str | None = None,
) -> MCPClient

Connect to a Streamable HTTP MCP endpoint. headers carries credentials (e.g. Authorization).

list_tools async

list_tools(*, refresh: bool = False) -> list[MCPToolInfo]

Every tool the server offers (all pages), cached until refresh=True.

call async

call(
    name: str, arguments: dict[str, Any] | None = None
) -> tuple[str, Any]

Call a tool; returns (text, structured). Raises ToolError if the server reports an error.

read_resource async

read_resource(uri: str) -> str

The text of a resource (binary parts are summarized, not returned).

load_tools async

load_tools(**kw: Any) -> list[Tool]

list_tools() then tools(...): the usual way to fill a registry.

tools

tools(
    *,
    prefix: str | None = None,
    include: Iterable[str] | None = None,
    exclude: Iterable[str] = (),
    approval: Approval = "none",
    trust_annotations: bool = True,
    timeout_s: float = 60.0,
    max_result_chars: int | None = 20000,
    infos: Sequence[MCPToolInfo] | None = None,
) -> list[Tool]

The server's tools as AAA tools (call list_tools() first, or use load_tools).

include/exclude are glob patterns on the server's tool names. approval="destructive" requires human approval for every tool that is not read-only (MCP: destructive unless stated otherwise); "always" for all. trust_annotations=False ignores the server's hints and treats every tool as non-repeatable (at_most_once) and destructive.

MCPToolInfo dataclass

MCPToolInfo(
    name: str,
    description: str = "",
    parameters: dict[str, Any] = (
        lambda: {"type": "object", "properties": {}}
    )(),
    title: str | None = None,
    read_only: bool | None = None,
    destructive: bool | None = None,
    idempotent: bool | None = None,
    open_world: bool | None = None,
)

A server tool, normalized across SDK versions.

is_destructive property

is_destructive: bool

Per the MCP spec: destructive unless read-only or explicitly marked non-destructive.

mcp_memory

mcp_memory(
    client: MCPClient,
    *,
    add: str,
    search: str,
    forget: str | None = None,
    delete: str | None = None,
    build_add: AddArgs = default_add_args,
    build_search: SearchArgs = default_search_args,
    build_forget: ForgetArgs = default_forget_args,
    build_delete: DeleteArgs = default_delete_args,
    parse: Parse = default_parse,
    idempotent: bool = False,
) -> CallableMemory

A Memory backed by the named tools of client. See the module docstring.

mcp_tool_name

mcp_tool_name(
    prefix: str, name: str, *, max_len: int = 64
) -> str

A model-API-safe tool name: [A-Za-z0-9_-], at most 64 characters, stable and unique.

render_content

render_content(blocks: Iterable[Any]) -> str

Flatten MCP content blocks into text a model can read. Binary content becomes a short marker (it would only burn context); fetch it yourself with read_resource if you need the bytes.

SchemaTool

SchemaTool(
    call: Callable[[dict[str, Any]], Any],
    *,
    name: str,
    description: str,
    parameters: dict[str, Any],
    **opts: Any,
)

Bases: Tool

A tool defined by an explicit JSON Schema and an async callable taking the argument dict, instead of a Python signature. This is how tools that arrive from outside Python (MCP servers, OpenAPI operations, ...) join a ToolRegistry; they get the same argument validation, approval, timeout and execution-guard behavior as @tool functions.

validate_arguments

validate_arguments(
    schema: dict[str, Any], arguments: dict[str, Any]
) -> list[str]

Check arguments against a JSON Schema; returns human-readable problems (empty = valid). Uses jsonschema when installed (it is, with the mcp extra); otherwise a conservative subset: required keys, unknown keys when additionalProperties is false, and primitive types.