AgentYard SDK

Python SDK for building, registering, and managing A2A agents on the AgentYard platform.

Installation

pip install agentyard

# From local source
pip install ./backend/sdk

agentyard.v2 — start here (current, recommended)

Agents are pure functions; the runtime — not the agent — owns transport, retries, validation, and secrets. Declare what the agent needs, don't wire it up by hand.

from pydantic import BaseModel
from agentyard.v2 import yard
from agentyard.v2.types import Resource

class Input(BaseModel):
    text: str

class Output(BaseModel):
    summary: str

@yard.agent(
    name="my-agent",
    namespace="acme/finance",
    intent="Summarize a block of text in one sentence.",
    inputs=Input,
    outputs=Output,
    needs=[Resource.llm(provider="bedrock")],
)
async def handler(input: Input, ctx) -> dict:
    summary = await ctx.llm(f"Summarize in one sentence: {input.text}")
    return {"summary": summary}

if __name__ == "__main__":
    yard.run()

Model / provider configuration

ctx.llm is a callable, not a method-only object — await ctx.llm("...") returns a str, and await ctx.llm("...", schema=SomeModel) returns a validated SomeModel instance (the LLM is instructed to emit matching JSON, which is code-fence-stripped and parsed before validation).

await ctx.llm("Summarize this in one sentence.")
await ctx.llm("Extract the fields.", schema=SomeModel, model="claude-haiku-4-5-20251001")

Provider resolution order: config["provider"] (K8s/mesh deploys read /yard/config.yaml) → YARD_LLM_PROVIDER env var → defaults to "anthropic". Supported providers: anthropic, bedrock, openai, cohere.

AWS Bedrock: set YARD_LLM_PROVIDER=bedrock — no API key required, it authenticates via the ambient AWS credential chain (instance role, env credentials, etc.), matching Bedrock's own auth model. Region resolves from config["bedrock_region"] → AWS_DEFAULT_REGION → AWS_REGION → "us-east-1". The default model is the cross-region inference profile ID us.anthropic.claude-haiku-4-5-20251001-v1:0 (Bedrock rejects the bare on-demand foundation-model ID for this model family — it must be the inference-profile form). Override the model via config["model"] or the YARD_LLM_MODEL env var.

@yard.agent fields (name, namespace, intent required; everything else optional): version (default "1.0.0"), inputs/outputs (Pydantic models — schemas auto-derive from these), is_idempotent/is_long_running/ is_pure (behavior hints), needs (list of Resource(...) — what the agent requires: Resource.llm(provider=...), Resource.postgres(...), Resource.redis(...), Resource.secrets([...]), Resource.s3(...)), memory (dict or MemoryContract(reads=[...], writes=[...], scope=...)), failure (FailurePolicy(mode=..., max_retries=..., ...)), port (default 9000), image (Docker image; defaults to agentyard-{name}:latest).

The rest of ctx (the "3-verb" runtime surface):

• ctx.memory — scratch (per-invocation dict, no persistence) plus ctx.memory.find(...) / ctx.memory.cite(...) for semantic recall + citation tracking. ACL-enforced against the agent's declared memory contract.
• ctx.invoke(agent=...) / ctx.invoke(system=...) / ctx.invoke(capability=...) / ctx.invoke(human=...) — dispatch to another agent, a whole sub-system, a capability-tagged agent, or a human-in-the-loop checkpoint. Exactly one of the four kwargs is set.

Self-registration is automatic on startup — no auto_register wiring needed, unlike v1. Disable with YARD_AUTO_REGISTER=false.

Full reference, including @yard.mcp_server, event hooks, and every deployment target: see SDK_GUIDE.md in the main repo.

---

Legacy v1 API (from agentyard import yard)

Kept for backward compatibility — decorator-based, imperative (ctx.emit/ctx.tracer/ctx.get_breaker). New agents should use agentyard.v2 above.

Quick Start

from agentyard import yard

@yard.agent(
    name="summarizer",
    namespace="default",
    description="Summarizes text input",
    version="1.0.0",
    framework="custom",
    input_schema={
        "type": "object",
        "properties": {"text": {"type": "string"}},
        "required": ["text"],
    },
    output_schema={
        "type": "object",
        "properties": {"summary": {"type": "string"}},
    },
)
async def summarize(input: dict) -> dict:
    text = input["text"]
    return {"summary": text[:200] + "..."}

# Start the agent (HTTP server on port 9000 by default)
yard.run()

Context Object

Agent functions can optionally accept a YardContext for access to shared memory, tools, progress streaming, and structured logging:

from agentyard import yard, YardContext

@yard.agent(name="smart-agent", ...)
async def handler(input: dict, ctx: YardContext = None) -> dict:
    # Read/write shared memory (Redis-backed in systems)
    prev = await ctx.memory.get("previous_output") if ctx else None
    if ctx:
        await ctx.memory.set("my_key", {"data": "value"})

    # Use MCP tools (sidecar discovery via YARD_MCP_TOOLS env)
    if ctx and ctx.tools:
        result = await ctx.tools.execute("search_code", {"q": "bug"})

    # Emit streaming progress events
    if ctx:
        await ctx.emit_progress({"status": "halfway", "pct": 50})

    # Structured logging (feeds into AgentYard monitoring)
    if ctx:
        ctx.log("Processing complete", level="info", tokens=150)

    return {"result": "done"}

Context is automatically injected by both HTTP and Redis Stream transports when the agent runs inside a system.

LLM client — ctx.llm.complete()

v1's ctx.llm is a richer, imperative client with automatic provider routing by model name (no per-provider SDK boilerplate), retry, streaming, semantic caching, and per-call cost tracking:

@yard.agent(name="doc-summarizer", namespace="acme/docs")
async def summarize(input: dict, ctx) -> dict:
    response = await ctx.llm.complete(
        prompt=f"Summarize this document in 3 bullets:\n\n{input['text']}",
        model="gpt-4o-mini",
        max_tokens=300,
        temperature=0.2,
    )
    return {"summary": response.text, "cost_usd": response.cost_usd}

Set OPENAI_API_KEY and/or ANTHROPIC_API_KEY at runtime; YARD_LLM_DEFAULT_MODEL sets the default model when model= is omitted (default gpt-4o-mini). Model name determines the provider automatically:

Model	Provider	Input $/1M	Output $/1M
gpt-4o	openai	2.50	10.00
gpt-4o-mini	openai	0.15	0.60
claude-3-5-sonnet	anthropic	3.00	15.00
claude-3-5-haiku	anthropic	0.80	4.00
claude-opus-4-6	anthropic	15.00	75.00

Unknown model names fall back to a heuristic (gpt-* / o1-* → OpenAI, claude* → Anthropic). LLMResponse carries text, model, provider, tokens_in/tokens_out, cost_usd, latency_ms, finish_reason, cached, and raw. Streaming (ctx.llm.stream(...)) is supported for OpenAI and Anthropic; Bedrock streaming is not yet implemented on this path. See LLMError/LLMRateLimitError/LLMProviderError for error handling and SDK_GUIDE.md for the full reference (caching, message lists, retry tuning).

Shared Memory

When agents run as nodes in a system, they share memory via Redis:

# Read a value set by a previous node
value = await ctx.memory.get("analysis_result")

# Write a value for downstream nodes
await ctx.memory.set("my_output", {"score": 0.95})

# Read all shared memory
all_data = await ctx.memory.get_all()

# Delete a key
await ctx.memory.delete("temp_key")

Memory strategies (set via YARD_MEMORY env var):

• shared_bus (default) — all nodes read/write freely
• isolated / none — writes are silently dropped

MCP Tools

Agents can call MCP tool servers deployed as sidecars:

from agentyard import ToolsClient

tools = ToolsClient()  # Reads YARD_MCP_TOOLS="github:3100,slack:3101"

# List available tools
all_tools = await tools.list_tools()
github_tools = await tools.list_tools(server="github")

# Execute a tool
result = await tools.execute("create_issue", {"title": "Bug", "body": "..."})
result = await tools.execute("send_message", {"channel": "#dev"}, server="slack")

Input/Output Validation

Schemas declared in @yard.agent() are validated automatically on every request:

@yard.agent(
    name="parser",
    input_schema={
        "type": "object",
        "properties": {
            "text": {"type": "string"},
            "max_length": {"type": "integer"},
        },
        "required": ["text"],
    },
    output_schema={
        "type": "object",
        "properties": {"parsed": {"type": "object"}},
    },
)
def parse(input: dict) -> dict:
    ...

Invalid input returns HTTP 400 with the validation error message.

Middleware (Before/After Hooks)

Register hooks that run before and after every invocation:

from agentyard import before, after, on_error

@before
def add_timestamp(input_data, ctx):
    input_data["_received_at"] = "2024-01-01T00:00:00Z"
    return input_data  # Return modified input

@after
def add_metadata(input_data, output, ctx):
    output["_version"] = "1.0"
    return output  # Return modified output

@on_error
def log_failure(input_data, error, ctx):
    print(f"Agent failed: {error}")

Hooks support both sync and async functions.

Metrics

Agent invocations are automatically recorded to Redis for AgentYard analytics:

• Total calls, success/error counts
• Cumulative and per-call latency
• Rolling window of last 1000 latencies

No configuration needed — metrics are collected automatically when YARD_REDIS_URL is set.

Structured Logging

from agentyard import get_logger

log = get_logger("my-agent")
log.info("Processing request", tokens=150, model="gpt-4")
log.warning("Slow response", latency_ms=5000)
log.error("Failed to call downstream", error="timeout")

Outputs JSON to stderr, compatible with AgentYard log collection:

{"ts": "2024-01-01T00:00:00Z", "level": "info", "agent": "my-agent", "msg": "Processing request", "tokens": 150}

Testing

Test agents locally without Docker, Redis, or any infrastructure:

from agentyard.testing import test_agent, AgentTestClient

# Quick test
result = test_agent(summarize, {"text": "Hello world"})
assert "summary" in result

# Test client with agent card inspection
client = AgentTestClient(summarize)
result = client.invoke({"text": "Hello"})
card = client.agent_card()
health = client.health()

Schema validation runs during tests by default. Disable with validate=False:

result = test_agent(handler, {"raw": "data"}, validate=False)

Transport Modes

Set YARD_TRANSPORT to choose how the agent receives traffic:

Value	Description
http (default)	FastAPI server with A2A endpoints
redis-stream	Redis Stream consumer (requires YARD_SYSTEM_ID + YARD_NODE_ID)
both	HTTP for health checks + Redis for production traffic

CLI Commands

agentyard publish

Register agents with the AgentYard registry:

agentyard publish -f my_agent.py
agentyard publish -m my_package.agent

agentyard build

Build a Docker image for an agent:

agentyard build -f my_agent.py
agentyard build -f my_agent.py -t myrepo/agent:1.0
agentyard build -f my_agent.py --push

agentyard list

agentyard list
agentyard list --namespace acme --framework langchain
agentyard list -q "invoice parser" --limit 10

agentyard info

agentyard info invoice-parser
agentyard info 550e8400-e29b-41d4-a716-446655440000

agentyard health

agentyard health invoice-parser

agentyard deprecate

agentyard deprecate 550e8400... --note "Replaced by v2"

agentyard stats

agentyard stats

agentyard config

agentyard config set registry-url http://localhost:8000
agentyard config set token ayard_tok_abc123
agentyard config get registry-url
agentyard config show

Environment Variables

Variable	Default	Description
YARD_TRANSPORT	http	Transport mode: http, redis-stream, both
YARD_PORT	9000	HTTP server port
YARD_REDIS_URL	redis://redis:6379	Redis connection URL
YARD_SYSTEM_ID		System ID (required for redis-stream)
YARD_NODE_ID		Node ID within system (required for redis-stream)
YARD_MEMORY	shared_bus	Memory strategy: shared_bus, isolated, none
YARD_MCP_TOOLS		MCP sidecar discovery: github:3100,slack:3101
YARD_LLM_PROVIDER	anthropic	v2 ctx.llm provider: anthropic, bedrock, openai, cohere
YARD_LLM_MODEL	(provider default)	v2 ctx.llm model override
YARD_LLM_DEFAULT_MODEL	gpt-4o-mini	v1 ctx.llm.complete() default model
YARD_AUTO_REGISTER	true	Disable self-registration on startup
YARD_AGENT_NAME		Agent name for logging
AGENTYARD_REGISTRY_URL	http://registry:8001	Registry URL for auto-registration
AGENTYARD_URL		Alternative registry URL

Architecture

@yard.agent decorator
    |
    v
yard.run() --> selects transport
    |
    +-- http_adapter.py --> FastAPI server
    |       - /.well-known/agent.json (A2A agent card)
    |       - POST / (process input)
    |       - GET /health
    |
    +-- redis_adapter.py --> Redis Stream consumer
            - Reads from yard:system:{id}:node:{id}:in
            - Writes to yard:system:{id}:node:{id}:out
            - Traces to yard:system:{id}:trace

Both adapters:
    - Create YardContext with memory, tools, logging
    - Run before/after middleware hooks
    - Validate input/output schemas
    - Record metrics to Redis