A2A Adapter

Convert any AI agent into an A2A Protocol server in 3 lines.

A Python SDK that makes any agent framework (n8n, LangGraph, CrewAI, LangChain, OpenClaw, Claude Code, Codex, Ollama, or a plain function) compatible with the A2A (Agent-to-Agent) Protocol.

from a2a_adapter import N8nAdapter, serve_agent

adapter = N8nAdapter(webhook_url="http://localhost:5678/webhook/agent")
serve_agent(adapter, port=9000)

That's it. Your agent is now A2A-compatible with auto-generated AgentCard, task management, and streaming support — all handled by the A2A SDK.

Features

• 3-line setup — import, create, serve
• Built-in adapters — including n8n, LangChain, LangGraph, CrewAI, OpenClaw, Claude Code, Codex, Ollama, and more
• Streaming — auto-detected for LangChain and LangGraph
• Auto AgentCard — generated from adapter metadata, served at /.well-known/agent.json
• SDK-First — delegates task management, SSE, push notifications to the A2A SDK
• Extensible — register_adapter() for third-party frameworks
• Minimal surface — implement invoke(), get a full A2A server

Installation

pip install a2a-adapter                # Core (includes n8n, callable)
pip install a2a-adapter[crewai]        # + CrewAI
pip install a2a-adapter[langchain]     # + LangChain
pip install a2a-adapter[langgraph]     # + LangGraph
pip install a2a-adapter[all]           # Everything

Quick Start

n8n Workflow

from a2a_adapter import N8nAdapter, serve_agent

adapter = N8nAdapter(webhook_url="http://localhost:5678/webhook/agent")
serve_agent(adapter, port=9000)

LangChain (with streaming)

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from a2a_adapter import LangChainAdapter, serve_agent

chain = ChatPromptTemplate.from_template("Answer: {input}") | ChatOpenAI(model="gpt-4o-mini")
adapter = LangChainAdapter(runnable=chain, input_key="input")
serve_agent(adapter, port=8002)  # Streaming auto-detected!

LangGraph (with streaming)

from a2a_adapter import LangGraphAdapter, serve_agent

graph = builder.compile()  # Your LangGraph workflow
adapter = LangGraphAdapter(graph=graph)
serve_agent(adapter, port=9002)

CrewAI

from a2a_adapter import CrewAIAdapter, serve_agent

adapter = CrewAIAdapter(crew=your_crew, timeout=600)
serve_agent(adapter, port=8001)

OpenClaw

from a2a_adapter import OpenClawAdapter, serve_agent

adapter = OpenClawAdapter(thinking="low", agent_id="main")
serve_agent(adapter, port=9008)

Ollama (local LLM)

from a2a_adapter import OllamaAdapter, serve_agent
from a2a_adapter.integrations.ollama import OllamaClient

client = OllamaClient(model="llama3.2")
adapter = OllamaAdapter(client=client, name="My Local LLM")
serve_agent(adapter, port=10010)

Claude Code

from a2a_adapter import ClaudeCodeAdapter, serve_agent

adapter = ClaudeCodeAdapter(working_dir="/path/to/project")
serve_agent(adapter, port=9010)

Security note: By default, skip_permissions is False. Without it
(and without a pre-configured Claude Code permissions file), tool-use
calls may not proceed in unattended mode. For trusted, sandboxed
environments only:

adapter = ClaudeCodeAdapter(working_dir="...", skip_permissions=True)

Or via env var: A2A_CLAUDE_SKIP_PERMISSIONS=1

Codex

from a2a_adapter import CodexAdapter, serve_agent

adapter = CodexAdapter(working_dir="/path/to/project")
serve_agent(adapter, port=9011)

Security note: By default, bypass_approvals and skip_git_check
are False. For trusted, sandboxed environments only:

adapter = CodexAdapter(working_dir="...", bypass_approvals=True, skip_git_check=True)

Or via env vars: A2A_CODEX_BYPASS_APPROVALS=1 and A2A_CODEX_SKIP_GIT_CHECK=1

Custom Function

from a2a_adapter import CallableAdapter, serve_agent

async def my_agent(inputs):
    return f"Echo: {inputs['message']}"

adapter = CallableAdapter(func=my_agent, name="Echo Agent")
serve_agent(adapter, port=9005)

Custom Adapter Class

For full control, subclass BaseA2AAdapter:

from a2a_adapter import BaseA2AAdapter, serve_agent

class MyAdapter(BaseA2AAdapter):
    async def invoke(self, user_input: str, context_id: str | None = None, **kwargs) -> str:
        return f"You said: {user_input}"

serve_agent(MyAdapter(), port=8003)

Architecture

A2A Caller (other agents)
    │  A2A Protocol (HTTP + JSON-RPC 2.0 / SSE)
    ▼
┌──────────────────────────────────────────────┐
│  A2A SDK (DefaultRequestHandler, TaskStore)  │  ← handles protocol
├──────────────────────────────────────────────┤
│  AdapterAgentExecutor (bridge layer)         │  ← adapts interface
├──────────────────────────────────────────────┤
│  Your Adapter (invoke / stream)              │  ← YOUR CODE HERE
├──────────────────────────────────────────────┤
│  Framework (n8n / LangChain / CrewAI / ...)  │
└──────────────────────────────────────────────┘

Design principle: Adapters answer ONE question — "given text, return text." Everything else (task management, SSE streaming, push notifications, AgentCard serving) is handled by the A2A SDK.

See ARCHITECTURE.md for detailed design documentation, and DESIGN_V0.2.md for the v0.2 design rationale.

API Reference

Core

BaseA2AAdapter (implement this)

Adapter Support

Input Handling

All adapters support a 3-priority input pipeline:

1. input_mapper (highest) — custom function (raw_input, context_id) -> dict
2. parse_json_input — auto-parse JSON strings to dict
3. input_key (fallback) — map text to {input_key: text}

Config-driven Loading

from a2a_adapter import load_adapter

adapter = load_adapter({
    "adapter": "n8n",
    "webhook_url": "http://localhost:5678/webhook/agent",
    "timeout": 60,
})

Third-party Adapters

from a2a_adapter import register_adapter, BaseA2AAdapter

@register_adapter("my_framework")
class MyFrameworkAdapter(BaseA2AAdapter):
    async def invoke(self, user_input, context_id=None, **kwargs):
        return "Hello from my framework!"

# Now loadable via config:
adapter = load_adapter({"adapter": "my_framework"})

Advanced: ASGI Deployment

For production deployments with Gunicorn/Hypercorn:

from a2a_adapter import N8nAdapter, to_a2a

adapter = N8nAdapter(webhook_url="http://localhost:5678/webhook/agent")
app = to_a2a(adapter)  # Returns Starlette ASGI app

# Deploy with: gunicorn app:app -k uvicorn.workers.UvicornWorker

Migration from v0.1

v0.2 is backwards compatible — v0.1 code still works but emits deprecation warnings.

Examples

The examples/ directory contains working examples for each adapter:

python examples/n8n_agent.py          # n8n
python examples/langchain_agent.py    # LangChain (streaming)
python examples/langgraph_server.py   # LangGraph (streaming)
python examples/crewai_agent.py       # CrewAI
python examples/openclaw_agent.py     # OpenClaw
python examples/ollama_agent.py         # Ollama (local LLM)
python examples/claude_code_agent.py   # Claude Code
python examples/codex_agent.py         # Codex
python examples/custom_adapter.py      # Custom BaseA2AAdapter
python examples/single_agent_client.py  # Test any running agent

See examples/README.md for details.

Testing

pip install a2a-adapter[dev]
pytest                    # All tests
pytest tests/unit/        # Unit tests only

Contributing

We welcome contributions! See CONTRIBUTING.md for guidelines.

Quick start:

1. Fork & clone
2. pip install -e ".[dev]"
3. Make changes + add tests
4. pytest to verify
5. Submit a PR

License

Apache-2.0 — see LICENSE.

Built with care by HYBRO AI. Powered by the A2A Protocol.