elsai

Appearance

A2A API ​

Agent-to-Agent protocol server and client types (A2A Protocol v1.0 via a2a-sdk>=1.0.0). For concepts, production examples, and migration from v0.3, see Agent-to-Agent (A2A).

Installation ​

bash
pip install --extra-index-url https://elsai-agents.elsai.ai/root/ "elsai-agents[a2a]==0.3.0"

A2AServer ​

python
from elsai.multiagent.a2a import A2AServer

Wraps a template agent or agent factory as an A2A-compatible HTTP server.

python
A2AServer(
    agent,                          # Agent instance or factory callable
    *,
    name=None,
    description=None,
    version="0.0.1",
    skills=None,
    **options,
)
ParameterTypeDefaultDescription
agentAgent | Callable[[A2AContextContext], Agent](required)Template agent (cloned per context) or factory that builds a fully configured agent per context
namestr | Noneagent.name (template mode)Display name in the agent card
descriptionstr | Noneagent.description (template mode)Description in the agent card
versionstr"0.0.1"Agent version string
skillslist[AgentSkill] | NoneNoneCapabilities advertised in the agent card

Network options ​

ParameterDefaultDescription
host127.0.0.1Bind address
port9000Port
http_urlauto-detectedPublic URL advertised in the agent card; path sets mount prefix
serve_at_rootFalseMount routes at / when http_url has a path prefix
public_url_schemeautohttp or https for auto-generated URLs

Session options ​

ParameterDefaultDescription
session_storeauto-createdPre-built A2ASessionStore; pass to share or customize the pool
session_manager_factoryfile-backed under ELSAI_A2A_SESSION_DIR(context_id) -> SessionManager for template-clone mode
max_active_contexts100LRU limit for in-memory context agents (None = unlimited)
strict_tool_policyFalseWhen True, reject startup if warnlisted tools are not PER_CONTEXT-safe

Handler options ​

ParameterDefaultDescription
task_storein-memoryPersistence backend for A2A task state
queue_managerNoneOptional task queue manager
push_config_storeNoneOptional push notification config store
push_senderNoneOptional push notification sender
enable_a2a_compliant_streamingTrueStream content via artifact updates (A2A v1 compliant)
enable_v0_3_compatFalseAccept legacy v0.3 JSON-RPC wire format

Methods and properties ​

MemberDescription
serve()Start the HTTP server (blocking)
to_fastapi_app()Return a FastAPI app for embedding
to_starlette_app()Return a Starlette app for embedding
public_agent_carda2a.types.AgentCard advertised at /.well-known/agent-card.json
session_storeThe A2ASessionStore backing per-context agents

A2AAgent ​

python
from elsai.agent import A2AAgent

Client wrapper for calling a remote A2A server.

python
A2AAgent(
    endpoint,
    *,
    name=None,
    description=None,
    timeout=300,
    client_config=None,
)
ParameterTypeDefaultDescription
endpointstr(required)Remote A2A server base URL
namestr | Nonefrom agent cardName when wrapping as a tool
descriptionstr | Nonefrom agent cardDescription when wrapping as a tool
timeoutint300Request timeout in seconds (when no custom httpx_client)
client_configClientConfig | NoneNonea2a.client.ClientConfig for auth, transport, and streaming

Deprecated

a2a_client_factory is deprecated — use client_config instead.

MethodDescription
__call__(prompt)Synchronous invocation
invoke_async(prompt)Async invocation
stream_async(prompt)Async streaming — yields a2a_stream events, then AgentResult
get_agent_card()Fetch and cache the remote agent card
as_tool()Wrap as an AgentTool for orchestrators

A2ASessionStore ​

python
from elsai.multiagent.a2a import A2ASessionStore

Maps A2A context_id values to dedicated agents with durable session backing.

python
A2ASessionStore(
    agent,                              # Agent template or factory
    *,
    session_manager_factory=None,
    max_active_contexts=100,
    strict_tool_policy=False,
)
ParameterTypeDefaultDescription
agentAgent | Callable[[A2AContextContext], Agent](required)Template to clone or factory callable
session_manager_factoryCallable[[str], SessionManager] | Nonefile-backed defaultReturns a SessionManager per context_id (template-clone mode only)
max_active_contextsint | None100LRU in-memory pool limit
strict_tool_policyboolFalseFail when warnlisted tools are not PER_CONTEXT-safe

Properties ​

PropertyDescription
uses_factoryTrue when agents are created via a factory callable
in_flight_contextsfrozenset of context_id values currently holding the execution lock
metricsA2ASessionStoreMetrics instance

Methods ​

MethodDescription
get_agent(ctx)Return the agent for ctx.context_id, creating or restoring from storage
get_lock(context_id)Per-context asyncio.Lock
hold_context(context_id)Async context manager — acquires lock, marks context in-flight (excluded from LRU eviction)
close(context_id)Remove agent from memory; skips in-flight contexts; does not delete persisted data
close_all()Remove all non-in-flight agents from memory
probe_strict_tool_policy()Validate factory-built agents at startup when strict_tool_policy=True

A2AContextContext ​

python
from elsai.multiagent.a2a import A2AContextContext

Per-conversation context passed to agent factories.

FieldTypeDescription
context_idstrA2A conversation identifier (maps to session_id)
task_idstr | NoneOptional A2A task ID
metadatadict | NoneOptional request metadata

A2ASessionStoreMetrics ​

python
from elsai.multiagent.a2a import A2ASessionStoreMetrics
FieldTypeDescription
cache_hitsintTimes a cached context agent was reused
cache_missesintTimes a new context agent was created
evictionsintTimes an agent was removed from the in-memory pool by LRU

Tool session policy ​

python
from elsai.types.tools import A2AToolSessionScope
from elsai.multiagent.a2a import A2A_STATEFUL_TOOL_WARNLIST

A2AToolSessionScope ​

ValueDescription
SHAREDSafe to share across contexts (default for stateless @tool functions)
CONTEXT_KEYEDMutable state keyed by invocation_state["a2a_context_id"]
PER_CONTEXTRequires clone_for_a2a_context(context_id)

A2A_STATEFUL_TOOL_WARNLIST ​

Built-in tool names inspected at A2A server startup:

python_repl, shell, use_computer, code_interpreter, browser, mcp_client, graph, workflow, mem0_memory

When strict_tool_policy=True, each warnlisted tool on the template agent must declare PER_CONTEXT scope and implement clone_for_a2a_context().

ElsaiA2AExecutor ​

Internal request handler that routes A2A tasks through A2ASessionStore. Configured automatically by A2AServer; injects a2a_context_id into tool invocation_state.

Stream normalizer exports ​

For custom clients built on a2a-sdk 1.x, Elsai converts v1 StreamResponse chunks into the legacy A2AResponse shape used by A2AAgent.stream_async:

python
from a2a.types import SendMessageRequest
from elsai.multiagent.a2a import A2AStreamNormalizer, iter_normalized_responses

request = SendMessageRequest(message=message)
async for event in iter_normalized_responses(client, request):
    ...  # Message or (Task, update_event) tuple

normalizer = A2AStreamNormalizer()
legacy_event = normalizer.normalize(stream_response_chunk)
ExportDescription
iter_normalized_responsesAsync generator: v1 StreamResponse → legacy A2AResponse events
A2AStreamNormalizerStateful single-chunk converter for custom clients

Environment variables ​

Session and tool state ​

VariableDefaultDescription
ELSAI_A2A_SESSION_DIR{tempdir}/elsai-a2a-sessionsRoot for A2A session files and on-disk per-context tool state
PYTHON_REPL_PERSISTENCE_DIR./repl_state/ (cwd)Standalone (non-A2A) persistence root for python_repl

Production deployment (convention) ​

Used in A2A concept — S3 production example:

VariableDescription
A2A_SESSIONS_BUCKETS3 bucket for A2A conversation history
A2A_SESSIONS_PREFIXS3 key prefix (e.g. a2a/contexts/)
AWS_REGIONAWS region for S3 and Bedrock tools
A2A_PUBLIC_URLPublic URL for A2AServer(http_url=…) in the agent card

Requires standard AWS credentials (env vars or IAM role) when using S3 session backing.

Copyright © 2026 elsai foundry.

Built with VitePress