Integrations: model APIs, protocols and importers

Export tools to model APIs and agent protocols (MCP, A2A, AG-UI); import tools from LangChain, OpenAPI and MCP. Each module imports its SDK lazily: install the extra first (Integrations).

agent_adapter.integrations.openai

Render tools for the OpenAI Chat Completions and Responses APIs as plain dicts.

INTEGRATION = OpenAIIntegration() module-attribute

OpenAIIntegration

Bases: IntegrationBase

Exports Chat Completions (default) or Responses tool dicts; the host runs the calls.

export_tool(spec, *, executor=None)

One tool dict; executor is unused because the host executes the model's call.

chat_tool(spec, *, strict=True)

One tool in Chat Completions format.

chat_tools(registry, *, strict=True)

Every registry tool exported to OpenAI, in Chat Completions format.

responses_tool(spec, *, strict=True)

One tool in Responses API format (flat, no function wrapper).

responses_tools(registry, *, strict=True)

Every registry tool exported to OpenAI, in Responses API format.

agent_adapter.integrations.gemini

Gemini: function declarations, genai automatic-function-calling proxies, call dispatch.

INTEGRATION = GeminiIntegration() module-attribute

GeminiIntegration

Bases: IntegrationBase

Exports function declaration dicts; the host runs the calls (run_function_calls).

export_tool(spec, *, executor=None)

One function declaration dict; executor is unused.

afc_callable(spec, *, executor=None)

An async function for genai's GenerateContentConfig(tools=[...]) that runs spec.

genai builds the declaration from the synthesised __signature__ (each property keeps the Gemini schema; optional ones carry their default, None when they have none) and __doc__. Calls go through the runner; genai turns any exception into an error reply.

Raises:

Type	Description
ToolUsageError	spec.targets excludes gemini.

afc_callables(registry, *, executor=None)

afc_callable for every registry tool exported to Gemini.

function_declaration(spec)

One tool as a Gemini function declaration dict.

function_declarations(registry)

Every registry tool exported to Gemini, as function declaration dicts.

genai_tool(registry)

A google.genai.types.Tool holding every registry tool exported to Gemini.

Raises:

Type	Description
MissingDependency	google-genai is not installed (pip install "agent-adapter[gemini]").

run_function_calls(calls, *, executor=None, ctx=None, registry=None) async

Run the model's function calls; one function-response types.Part per call, in order.

calls holds types.FunctionCalls, parts with a function_call, or {"name", "args", "id"} mappings; a call without an id gets a stable one (name, arguments, and which occurrence of that same call it is), so a repeated or reordered turn gets the same ids. Tools are looked up in registry, else the executor's registry. An Executor runs the calls concurrently in one run; another runner runs them in order. Model-visible errors become {"error": {"kind", "message"}} responses; any other exception (ApprovalPending, ToolNotFound, ToolTimeout, ...) propagates.

ctx=None uses current_context() when one is set (inside an adapter run), else a fresh run per batch. For approval, pass a ctx with a stable run_id and repeat the same calls after the decision: calls that completed in a batch stopped by ApprovalPending are answered from memory on the retry, not run again.

Raises:

Type	Description
ToolUsageError	no registry, or an item is not a function call.
MissingDependency	google-genai is not installed.

agent_adapter.integrations.anthropic

Render tools for the Anthropic Messages API as plain dicts.

INTEGRATION = AnthropicIntegration() module-attribute

AnthropicIntegration

Bases: IntegrationBase

Exports Messages API tool dicts; the host runs the model's tool_use blocks.

export_tool(spec, *, executor=None)

One tool dict; executor is unused.

anthropic_tool(spec)

One tool in Anthropic Messages format.

anthropic_tools(registry)

Every registry tool exported to Anthropic, in Messages format.

tool_result_blocks(results)

tool_result content blocks answering the model's tool_use blocks, in order.

agent_adapter.integrations.langchain

LangChain: registry tools as StructuredTools, and LangChain tools imported as ToolSpecs.

INTEGRATION = LangChainIntegration() module-attribute

LangChainIntegration

Bases: IntegrationBase

Exports StructuredTools; the executor's approval port decides confirm tools.

export_tool(spec, *, executor=None)

One StructuredTool (see structured_tool).

import_langchain(registry, tools, *, prefix='', **policy)

Add LangChain BaseTools to registry (all or nothing) and return their specs.

The schema is what LangChain sends a model (convert_to_openai_function), so InjectedToolArg parameters stay out; the importer does not supply them. Names are sanitised to [A-Za-z0-9_]. A ToolException becomes ToolError; arguments the tool's own argument model rejects become ToolArgsInvalid (checked before the call).

Raises:

Type	Description
ToolDefinitionError	an item is not a BaseTool, or a bad name or option.
DuplicateTool	a name is already registered.
MissingDependency	langchain-core is not installed.

structured_tool(spec, *, executor=None)

spec as a LangChain StructuredTool that runs through the executor.

The model sees spec.json_schema() unchanged and validation is the spec's own. Model-visible errors raise ToolException (handle_tool_error=True turns it into an error ToolMessage); other exceptions propagate. Sync invoke uses run_sync. The run id comes from current_context() inside an adapter run, else from the RunnableConfig: configurable["run_id"], else configurable["thread_id"] (LangGraph's); without either, every call is a fresh run (a deferred approval cannot resume).

Raises:

Type	Description
ToolUsageError	spec.targets excludes langchain.
MissingDependency	langchain-core is not installed.

structured_tools(registry, *, executor=None)

structured_tool for every registry tool exported to LangChain.

agent_adapter.integrations.openapi

OpenAPI 3.x importer: one tool per operation, called over httpx.

INTEGRATION = OpenAPIIntegration() module-attribute

Auth

Bases: ABC

Adds credentials to one request's headers or query parameters.

apply(headers, params) abstractmethod async

Add the credentials in place.

BearerAuth

Bases: Auth

Authorization: Bearer <token>; a sync or async token callable is asked per call.

apply(headers, params) async

Set the Authorization header.

Raises:

Type	Description
ConfigError	the token callable returned no non-empty string.

HeaderAuth

Bases: _NamedValueAuth

A fixed header, e.g. HeaderAuth("X-API-Key", key).

apply(headers, params) async

Set the header.

OpenAPIIntegration

Bases: IntegrationBase

Imports tools (import_openapi); it has no tool export.

export_tool(spec, *, executor=None)

Not supported: OpenAPI is an import source.

Raises:

Type	Description
IntegrationError	always; the HTTP tool server publishes /openapi.json.

QueryAuth

Bases: _NamedValueAuth

A fixed query parameter, e.g. QueryAuth("api_key", key); prefer HeaderAuth.

The value is masked in this process's httpx log lines, but servers, proxies and their access logs still record query strings. Use it only when the API accepts nothing else.

apply(headers, params) async

Set the query parameter.

import_openapi(registry, spec, *, base_url=None, auth=None, headers=None, operations=None, prefix='', timeout=30.0, client_factory=None, **policy)

Add one tool per operation of an OpenAPI 3.x document (all or nothing).

spec is a mapping, JSON/YAML text or a file path. client_factory(base_url=, headers=, timeout=) makes the httpx.AsyncClient (default: httpx.AsyncClient), once per event loop. 4xx answers (except 429) raise ToolError; 429 and 5xx raise httpx.HTTPStatusError, which the executor retries (retries=) before ToolFailed.

Raises:

Type	Description
ConfigError	bad document, options, server URL or unknown operations.
DuplicateTool	a tool name is already registered.
MissingDependency	httpx (or pyyaml for YAML) is not installed.

agent_adapter.integrations.openapi.auth

agent_adapter.integrations.openapi.auth re-exports, documented under the canonical path: Auth, BearerAuth, HeaderAuth, QueryAuth.

agent_adapter.integrations.mcp

MCP: serve registry tools to any MCP client, and import tools from MCP servers.

APPROVAL_ID_META = f'{_META_PREFIX}approval_id' module-attribute

CALL_ID_META = f'{_META_PREFIX}call_id' module-attribute

ERROR_KIND_META = f'{_META_PREFIX}error_kind' module-attribute

INTEGRATION = MCPIntegration() module-attribute

RUN_ID_META = f'{_META_PREFIX}run_id' module-attribute

TASKS_EXTENSION = 'io.modelcontextprotocol/tasks' module-attribute

TASK_ID_META = 'agent-adapter/task_id' module-attribute

MCPIntegration

Bases: IntegrationBase

Exports mcp.types.Tool definitions; mcp_server runs the calls.

bind_context(native)

A context with the client's raw, unscoped _meta keys (for inspection only).

Never build a server on these ids: mcp_server scopes them by principal and honours a call key only to resume a pending result.

export_tool(spec, *, executor=None)

The MCP tool definition; executor is used by mcp_server, not here.

MemoryTaskStore

In-process TaskStore: shared by the servers of one process only.

Past max_entries the oldest entries go first (WARNING).

MemoryTokenStorage

Tokens and client registration for one server, in this process only; never logged.

RedisTaskStore

TaskStore on Redis (agent-adapter[redis]), shared by every worker and host.

client_factory() returns a redis.asyncio.Redis (one per event loop), e.g. lambda: Redis.from_url(url). States are SET PX JSON values under prefix. Backend errors propagate; a task then stays readable only on its own worker.

aclose() async

Close the running loop's client.

TaskOptions dataclass

How mcp_server(tasks=...) answers 2026-07-28 clients that accept tasks.

Parameters:

Name	Type	Description	Default
threshold_s	float | None	A call still running after this many seconds becomes a task; None makes tasks only of calls that wait for approval.	5.0
ttl_s	float	How long a task is kept after it is created.	3600.0
poll_interval_s	float	The poll interval suggested to clients, and how often a task waiting for approval re-reads the approval store.	2.0
approval_url	str | None	An https page where a person decides the approval, with {approval_id} in it; clients with URL-mode elicitation are sent there.	None
max_tasks	int	Tasks kept at once; past it, calls are answered without a task.	10000
max_per_principal	int	Tasks kept at once for one caller.	100
store	TaskStore | None	Shared task state (e.g. RedisTaskStore) so any worker answers tasks/*; None keeps tasks in this process.	None

Raises:

Type	Description
ConfigError	a value out of range, or an approval_url that is not https (or http on a loopback host) or lacks {approval_id}; a store that is not a TaskStore.

approval_link(approval_id)

The approval page for approval_id (the id URL-quoted), or None without one.

TaskStore

Bases: Protocol

Task states (written by the worker running the task) and signals (from any worker).

delete(task_id) async

Forget the state.

flag(task_id, name, ttl_s) async

Raise the signal name (e.g. a cancel request) for ttl_s seconds.

flagged(task_id, names) async

Which of names are raised.

get(task_id) async

The live state, or None.

put(task_id, state, ttl_s) async

Store state for ttl_s seconds, replacing the previous one.

client_metadata_document(client_metadata_url, *, redirect_uris, client_name='agent-adapter', scope=None)

The JSON to publish at client_metadata_url; its client_id is that URL.

Raises:

Type	Description
ConfigError	the URL is not https with a path, or a redirect URI is missing or unsafe.

import_mcp(registry, client, *, prefix='', names=None, send_ids=False, **policy) async

Register the tools of an open mcp.Client (all of them, or the remote names).

The tools call the server through client, so they work only while it stays open.

Parameters:

Name	Type	Description	Default
prefix	str	Prepended to each local name; characters other than letters, digits and _ become _.	''
send_ids	bool	Send the local run id as the remote run key (servers you trust). The resume keys of a remote pending approval are always sent back on the retry.	False
**policy	Any	Execution options as for build_spec; they override the defaults kind="mcp", tags={"mcp"} and the effects / idempotent the remote annotations imply.	{}

Raises:

Type	Description
IntegrationError	a requested name is not on the server, or listing never ends.
(ToolDefinitionError, DuplicateTool, SchemaNotPortable)	as for registry.add_many; nothing is registered then.

mcp_client(target, *, headers=None, auth=None, tasks=False, task_wait_s=DEFAULT_TASK_WAIT_S)

An unopened mcp.Client for an http(s) URL (Streamable HTTP), a StdioServerParameters, an in-process Server / MCPServer, or a client transport.

Parameters:

Name	Type	Description	Default
auth	Auth | None	An httpx2.Auth for a URL, e.g. oauth_client(...); https only, except on a loopback host.	None
tasks	bool	Declare the 2026-07-28 tasks extension; a remote task is followed for up to task_wait_s seconds, then the call fails with its task_id and can resume it.	False

Raises:

Type	Description
ToolUsageError	a non-URL string, headers / auth for a non-URL target, an Authorization header with auth, or another type.
ConfigError	auth over plain http to a non-loopback host; a bad task_wait_s.
MissingDependency	mcp is not installed.

mcp_server(registry, *, executor=None, name='agent-adapter', version=__version__, instructions=None, client_approval=False, request_state_keys=None, principal=None, allow_anonymous=None, approval_store=None, tasks=False)

A lowlevel MCP Server listing and calling registry.for_target("mcp").

Run and call ids are scoped by the caller's principal. Each option is described in docs/integrations/mcp.md (Security and Client approval); tasks (the 2026-07-28 tasks extension) in docs/integrations/mcp-tasks.md.

Raises:

Type	Description
ToolUsageError	bad registry or principal, or keys without client_approval.
ConfigError	bad request_state_keys, approval_store or tasks (see the docs).
MissingDependency	mcp is not installed.

mcp_tool(spec)

The MCP tool definition for spec: schemas, wrapped output schema, annotations.

oauth_client(server_url, *, client_metadata_url, redirect_uri, redirect_handler, callback_handler, storage=None, client_name='agent-adapter', scope=None)

An httpx2.Auth for mcp_client(server_url, auth=...): authorization code + PKCE.

The SDK discovers the protected-resource and authorization-server metadata, uses the Client ID Metadata Document at client_metadata_url when the server supports it (else dynamic registration), refreshes tokens, and binds stored credentials to their issuer.

Parameters:

Name	Type	Description	Default
redirect_handler	Callable[[str], Awaitable[None]]	Sends the person to the authorization URL.	required
callback_handler	Callable[[], Awaitable[AuthorizationCodeResult]]	Waits for the redirect and returns its code and state.	required
storage	Any	A TokenStorage; default MemoryTokenStorage() (per process).	None

Raises:

Type	Description
ConfigError	server_url not https (http only on a loopback host), or a bad client_metadata_url / redirect_uri.
MissingDependency	mcp is not installed.

output_schema_for(spec)

outputSchema: the result schema, wrapped under result unless it is an object.

streamable_http_app(registry, *, executor=None, path='/mcp', stateless=True, json_response=True, name='agent-adapter', version=__version__, instructions=None, client_approval=False, request_state_keys=None, principal=AUTHENTICATED, allow_anonymous=False, approval_store=None, tasks=False, **http_options)

An ASGI (Starlette) app serving mcp_server(registry, ...) over Streamable HTTP at path.

principal defaults to the SDK's bearer-token identity (client id, issuer, subject); without one, client keys are ignored unless allow_anonymous=True. http_options go to Server.streamable_http_app (host=, transport_security=, auth=, token_verifier=, ...). The app's lifespan must run (uvicorn does).

Without auth (auth= + token_verifier=) any client that reaches the port can call every tool: a non-local host then needs allow_anonymous=True, and on a local host requests from a non-loopback peer are refused (403). A custom principal is not auth; unless allow_anonymous=True, a request it names no caller for is refused.

Raises:

Type	Description
ConfigError	an http_options key this function sets; a non-local host without transport_security, or without auth or allow_anonymous=True; auth= or token_verifier= alone.
MissingDependency	mcp is not installed.

tool_annotations(spec)

MCP ToolAnnotations fields (snake_case) from effects, confirm and idempotent.

agent_adapter.integrations.mcp.annotations

agent_adapter.integrations.mcp.annotations re-exports, documented under the canonical path: APPROVAL_ID_META, CALL_ID_META, ERROR_KIND_META, RUN_ID_META, output_schema_for, tool_annotations.

agent_adapter.integrations.mcp.client

agent_adapter.integrations.mcp.client re-exports, documented under the canonical path: import_mcp, mcp_client.

agent_adapter.integrations.mcp.oauth

agent_adapter.integrations.mcp.oauth re-exports, documented under the canonical path: MemoryTokenStorage, client_metadata_document, oauth_client.

agent_adapter.integrations.mcp.server

Serve registry tools over MCP: a v2 lowlevel Server and a stateless Streamable HTTP app.

AUTHENTICATED = _Authenticated() module-attribute

TARGET = 'mcp' module-attribute

require_mcp()

Raise MissingDependency when the mcp package is not installed.

Also exported here, documented under the canonical path: mcp_server, mcp_tool, streamable_http_app.

agent_adapter.integrations.mcp.tasks

MCP tasks extension options: which slow or pending calls become task handles.

APPROVAL_URL_PLACEHOLDER = '{approval_id}' module-attribute

Also exported here, documented under the canonical path: MemoryTaskStore, RedisTaskStore, TASKS_EXTENSION, TASK_ID_META, TaskOptions, TaskStore.

agent_adapter.integrations.a2a

A2A: serve registry tools as an A2A agent, and import a remote agent's skills as tools.

CARD_PATH = '/.well-known/agent-card.json' module-attribute

INTEGRATION = A2AIntegration() module-attribute

PRINCIPAL_STATE = 'aa.a2a.principal' module-attribute

TOOLS_EXTENSION = 'urn:agent-adapter:a2a:tools:v1' module-attribute

A2AConnection

An open A2A client, the agent card it was built from, and its HTTP client.

aclose() async

Close the A2A client and its HTTP client.

A2AIntegration

Bases: IntegrationBase

Exports one AgentSkill per tool; a2a_app / a2a_executor run the calls.

export_tool(spec, *, executor=None)

The skill for spec; executor is used by a2a_app, not here.

Caller dataclass

A minimal principal: id scopes tasks, runs and approvals ("" is anonymous).

HostRefused

Bases: IntegrationError

An outbound A2A request to a host the guard does not allow.

ResponseTooLarge

Bases: IntegrationError

A remote A2A response over the byte limit.

ToolAgent

Reads a {tool, arguments} data part, runs it, and reports through TaskUpdater.

A pending approval parks the call server-side and moves the task to input-required. The next message on that task re-runs the parked call: the executor's port decides, or, with client_approval, an explicit {approval_id, approved} answer becomes a one-use token bound to the run, task id, tool and argument digest.

cancel(context, event_queue) async

Forget the parked call and mark the task cancelled.

execute(context, event_queue) async

Run the call in context.message, or resume the task's parked call.

a2a_app(registry, *, url, executor=None, authenticate=None, authorize=None, allow_anonymous=False, host='127.0.0.1', allowed_hosts=None, public_card=False, client_approval=False, name='agent-adapter', description='Tools served by agent-adapter.', version=__version__, max_body_bytes=MAX_BODY_BYTES, max_data_bytes=MAX_DATA_BYTES, task_store=None)

A Starlette app: the agent card at CARD_PATH, JSON-RPC A2A at /.

url is the public JSON-RPC URL written in the card. Without authenticate any client that reaches the port can call every tool: a non-local host then needs allow_anonymous=True, and a local one serves loopback peers only. Tasks, runs and approvals are scoped by the principal's id. task_store (an SDK TaskStore) defaults to bounded_task_store(). Options: docs/integrations/a2a.md.

Raises:

Type	Description
ConfigError	a non-local host without allowed_hosts, or without auth or allow_anonymous=True; bad limits or url.
MissingDependency	a2a-sdk (with its http-server extra) is not installed.

a2a_executor(registry, *, executor=None, client_approval=False, authorize=None, max_data_bytes=MAX_DATA_BYTES)

An a2a-sdk AgentExecutor running registry.for_target("a2a") tools.

Raises:

Type	Description
MissingDependency	a2a-sdk is not installed.

agent_card(registry, *, url, name='agent-adapter', description='Tools served by agent-adapter.', version=__version__)

An AgentCard for registry.for_target("a2a"), served at url over JSON-RPC.

The TOOLS_EXTENSION extension carries each skill's JSON input schema, so import_a2a can rebuild typed tools. Streaming and push notifications are off.

Raises:

Type	Description
ConfigError	url is not http(s), or more than MAX_SKILLS tools.
MissingDependency	a2a-sdk is not installed.

agent_skill(spec)

The skill for one tool: id and name are the tool name.

Raises:

Type	Description
ToolUsageError	spec is not exported to a2a.
MissingDependency	a2a-sdk is not installed.

bounded_task_store(*, ttl=TASK_TTL_S, max_tasks=MAX_TASKS, max_per_owner=MAX_TASKS_PER_OWNER, clock=time.monotonic)

An SDK TaskStore in memory: tasks expire ttl s after their last update.

Over a cap the oldest finished task goes first (an owner's cap only drops its own tasks). A rejected task keeps its status but not its message history.

Raises:

Type	Description
ValueError	a cap or ttl is not positive.

call_remote(connection, skill_id, parts, *, timeout_s=TASK_TIMEOUT_S, poll_interval_s=POLL_INTERVAL_S) async

Send one message, poll its task to a final or interrupted state, return its result.

Raises:

Type	Description
ToolError	the remote task failed, needs input or auth, timed out, or the agent is unreachable or answered too much.
HostRefused	a request went to a host the guard does not allow.

connect_a2a(url, *, headers=None, auth=None, allow_private=False, trusted_hosts=(), allow_cross_origin=False, timeout_s=TIMEOUT_S, max_response_bytes=MAX_RESPONSE_BYTES, http_transport=None) async

Fetch the agent card at url + CARD_PATH and open a non-streaming A2A client.

Every request (card, calls, interfaces the card names) passes the host guard: no private or local addresses unless allow_private or the host is in trusted_hosts. Sockets go only to the addresses that check saw; a custom http_transport (tests, ASGI) is checked per request but connects on its own. headers / auth (an httpx.Auth) go only to url's origin unless allow_cross_origin=True. Responses are capped at max_response_bytes.

Raises:

Type	Description
HostRefused	url (or an interface host) is not allowed.
IntegrationError	the card cannot be fetched or parsed, or names no usable interface.
MissingDependency	a2a-sdk is not installed.

import_a2a(registry, connection, *, prefix='', names=None, task_timeout_s=TASK_TIMEOUT_S, poll_interval_s=POLL_INTERVAL_S, **policy) async

Register one tool per remote skill (all, or the skill ids in names).

A skill the card's TOOLS_EXTENSION describes takes its typed arguments and is sent as {tool, arguments}; any other skill takes {"message": str} sent as text. The tools work while connection stays open. A remote input-required is a ToolError: it never approves anything here.

Parameters:

Name	Type	Description	Default
**policy	Any	Options as for build_spec; they override kind="a2a", tags={"a2a"}.	{}

Raises:

Type	Description
IntegrationError	a requested skill is missing, or the card has too many skills.
(ToolDefinitionError, DuplicateTool, SchemaNotPortable)	as for registry.add_many.

static_bearer(tokens)

An authenticator for Authorization: Bearer <token> over a {token: caller id} table.

Raises:

Type	Description
ToolUsageError	an empty token or caller id.

agent_adapter.integrations.ag_ui

AG-UI 1.0: stream runs as AG-UI events; approvals become interrupts, resumes ask the port.

Agent = Callable[[AGUIRun], Awaitable[Any]] module-attribute

Authenticate = Callable[[Any], Awaitable[str | None]] module-attribute

INTEGRATION = AGUIIntegration() module-attribute

TARGET = 'ag_ui' module-attribute

AGUIBridge

Streams AG-UI runs of agent functions over one executor; holds the open interrupts.

__init__(executor, *, port=None, interrupts=None, max_events=MAX_EVENTS, trace=True)

port decides resumed calls (default: the executor's own port).

Resumed calls are looked up in executor.registry, the registry the executor runs (the default registry for an Executor() built without one).

sse(run_input, agent, *, principal='') async

stream encoded as text/event-stream frames.

stream(run_input, agent, *, principal='') async

The run's AG-UI events, from RUN_STARTED to RUN_FINISHED or RUN_ERROR.

principal is the authenticated caller; only they may answer this run's interrupts. Closing the iterator early (a client disconnect) cancels the agent.

Raises:

Type	Description
AGUIInputInvalid	run_input is invalid (before any event).

AGUIInputInvalid

Bases: ToolUsageError

A RunAgentInput that is not valid JSON, fails the schema, or exceeds a bound.

AGUIIntegration

Bases: IntegrationBase

Describes registry tools as AG-UI Tools; runs and approvals go through AGUIBridge.

export_tool(spec, *, executor=None)

An AG-UI Tool (name, description, JSON schema); AG-UI runs tools server side.

AGUIRun

One AG-UI run: its input, an executor and context bound to its event stream.

Run tools with call or pass executor and ctx to run_tool_loop: either way each call streams as TOOL_CALL_START/ARGS/END and TOOL_CALL_RESULT, and a call that waits for approval ends the run with an interrupt. Built by AGUIBridge.stream.

run_id property

The AG-UI run id the client sent (the executor's run id is ctx.run_id).

thread_id property

The AG-UI thread (conversation) id.

call(name, arguments=None, *, call_id='') async

Run one registry tool through the executor; every failure is a ToolResult.

custom(name, value)

Send an application CUSTOM event (value must be JSON).

messages()

The input's conversation as adapter messages (see to_messages).

text(content, *, role='assistant')

Stream one complete text message; returns its message id.

InterruptTable

The bridge's open interrupts, each answerable once, within its TTL, by its owner.

expiry()

The wall-clock time an interrupt opened now stops being answerable.

open(pending)

Record pending; over the owner's cap or a full table, the oldest go (fail closed).

take(interrupt_id, *, thread_id, principal)

Remove and return the open interrupt with this id, thread and principal.

Anything else (unknown, expired, answered before, another thread or principal) gives None and leaves the table unchanged.

PendingCall dataclass

One paused tool call, bound to its thread, the principal who started it and our run.

interrupt()

The AG-UI Interrupt a client answers with a resume entry.

agui_endpoint(bridge, agent, *, authenticate, max_body_bytes=MAX_BODY_BYTES, allowed_hosts=None)

A Starlette endpoint: authenticate, read a bounded JSON body, stream the run as SSE.

Mount it with Route("/agui", endpoint, methods=["POST"]), or on FastAPI with app.add_route("/agui", endpoint, methods=["POST"]). authenticate is required: pass anonymous only for a local, single-user app.

allowed_hosts lists the Host names served ("*": any). By default an anonymous endpoint serves local names only (DNS rebinding); any other endpoint serves every host.

Responses: 421 (Host not served), 401 (refused), 413 (body too large), 415 (not JSON), 422 (invalid input), else 200 text/event-stream.

anonymous(request) async

An Authenticate that lets everyone in as one anonymous principal (local use only).

parse_input(body, *, max_bytes=MAX_BODY_BYTES)

A validated, bounded RunAgentInput from a request body or a parsed mapping.

Raises:

Type	Description
AGUIInputInvalid	bad JSON, a schema error, or a bound exceeded (body size, id length, message or resume count, duplicate interrupt ids).

to_messages(run_input)

The input's conversation as adapter Messages; text parts only, other roles skipped.

Developer messages become system messages. Activity and reasoning messages are not sent to a model, so they are dropped.

Raises:

Type	Description
AGUIInputInvalid	an assistant tool call whose arguments are not a JSON object.