any-agent2agent 0.0.2

Toolkit for normalizing Agent Protocol (Agent2Agent) communication.

any_agent2agent

any_agent2agent now focuses on a single goal: normalize Agent2Agent (A2A) / Agent Protocol communication so any application can talk to any remote agent without learning a custom payload shape.

What's included

• AgentProtocolAdapter – HTTP client that speaks /v1/exchange in the Agent Protocol format.
• Server helpers – create_lambda_handler, create_fastapi_router, and friends with protocol="agent_protocol" so you can expose an agent quickly. FastAPI/Flask helpers auto-register themselves on an existing app when detected, so you rarely need to spin up a new server.
• Bedrock tool helper – create_bedrock_tool_handler converts any agent into an AWS Bedrock Agent tool/action group payload.
• Shared models – AgentMessage, BaseAgent, and a registry-backed AgentFactory so future adapters can be re‑added without redesigning the package.
• Schema registry loader – fetch schema manifests from GitHub/S3/local paths and auto-wire schema adapters for apps that need bespoke payloads.
• Secrets-aware config – ConfigManager fetches configuration values from AWS Secrets Manager or environment variables, used by the examples to drive ports and API keys.
• Lean tests & workflows – only the pieces that validate the Agent Protocol surface remain.

If you need to add more adapters later, the factory, base classes, and docs under docs/ADDING_PROVIDERS.md describe how to plug them back in.

Installation

pip install .
# or
pip install -e .[dev]

Quick start

Call a remote Agent Protocol service

from any_agent2agent import AgentFactory

agent = AgentFactory.create(
    "agent_protocol",
    base_url="https://agent.example.com",
    api_key="optional-bearer-token",
)

response = agent.send("Draft a Jira rollout plan")
print(response.content)
print(response.metadata)

Expose your agent with Agent Protocol schema

from typing import List

from fastapi import FastAPI

from any_agent2agent import AgentMessage
from any_agent2agent.server import (
    create_fastapi_router,
    create_lambda_handler,
)

def integrations_agent(messages: List[AgentMessage]) -> str:
    prompt = messages[-1].content
    return f"Answering integrations question: {prompt}"

lambda_handler = create_lambda_handler(
    integrations_agent,
    protocol="agent_protocol",
    default_metadata={"agent": "integrations"},
)

# FastAPI auto-registration detects the existing `app` variable.
app = FastAPI()
router = create_fastapi_router(
    integrations_agent,
    protocol="agent_protocol",
    route_path="/agent/exchange",
)
if not getattr(router, "_a2a_auto_registered", False):
    app.include_router(router)

The same protocol="agent_protocol" flag works with the FastAPI and Flask helpers so you can expose /agent/exchange over HTTP.

Need streaming output? Enable it on the server helper and hit /stream:

router = create_fastapi_router(
    integrations_agent,
    protocol="agent_protocol",
    streaming=True,
)

# POST /agent/stream returns newline-delimited JSON chunks
# Example client flag:
# {"messages": [...], "stream": true}

Lambda handlers can also opt-in:

lambda_handler = create_lambda_handler(
    integrations_agent,
    protocol="agent_protocol",
    streaming=True,
)

stream_response = lambda_handler({"messages": [...], "stream": True}, None)

Wire agents into Bedrock tools

from any_agent2agent import AgentMessage, create_bedrock_tool_handler

def integrations_agent(messages: list[AgentMessage]):
    yield "Checking integrations..."
    yield ("Slack is ready ✅", {"agent": "integrations"})

bedrock_handler = create_bedrock_tool_handler(
    integrations_agent,
    protocol="agent_protocol",
)

# Bedrock Lambda/action-group event
event = {
    "sessionId": "abc",
    "inputText": "Enable Slack",
    "stream": True,
}
response = bedrock_handler(event, None)

See docs/BEDROCK_TOOLING.md for the full contract plus configuration tips.

Bedrock bridge Lambda example

examples/bedrock_bridge_lambda.py demonstrates a deployable Lambda tool that receives Bedrock events, fans out to the flight and hotel agents via Agent Protocol, and returns a single combined message. Environment variables point the bridge at any Agent Protocol service (FLIGHT_AGENT_BASE_URL, HOTEL_AGENT_BASE_URL, etc.).

flowchart LR
    Bedrock[Bedrock Agent] -->|inputText| BridgeLambda["Bedrock bridge Lambda"]
    BridgeLambda -->|Agent Protocol| FlightAgent["Flight agent"]
    BridgeLambda -->|Agent Protocol| HotelAgent["Hotel agent"]
    FlightAgent -->|messages| BridgeLambda
    HotelAgent -->|messages| BridgeLambda
    BridgeLambda -->|responseBody.TEXT| Bedrock

Copy the example, zip it with your dependencies, and register the Lambda as an action group in Bedrock; the helper automatically formats the Bedrock tool response and surface streaming chunks when stream=true.

Reuse an existing FastAPI/Flask server

When the SDK spots a module-level app/application (or you explicitly provide one), it auto-registers the Agent Protocol endpoint:

• create_fastapi_router(...) calls include_router(...).
• create_flask_handler(...) calls add_url_rule(...).

Environment overrides:

• A2A_FASTAPI_APP=package.module:app forces a specific FastAPI app.
• A2A_FLASK_APP=package.module:application forces a specific Flask app.
• Set either variable to skip, off, 0, etc. to disable auto-detection.

If no app is found the helpers still return the router/handler, so you can register them manually (as before).

Configure the Agent Protocol server port

The examples/expose_agent_agentprotocol.py FastAPI app reads its port from AGENT_PROTOCOL_PORT. Values come from AWS Secrets Manager when AGENT_PROTOCOL_SECRET_NAME is set, otherwise they fall back to local environment variables:

export AGENT_PROTOCOL_SECRET_NAME="a2a/server/config"
# secret JSON must include {"AGENT_PROTOCOL_PORT": "8005"}
python examples/expose_agent_agentprotocol.py

ConfigManager drives this behavior and is available for your own settings as well:

from any_agent2agent.config import ConfigManager

config = ConfigManager(secret_name="a2a/server/config")
api_key = config.get("AGENT_PROTOCOL_API_KEY")

Multiple agents, same protocol

Instantiate as many AgentProtocolAdapter clients as you need—point them at different base_url values or reuse the same remote agent from different apps. AgentFactory keeps the registration hooks required for future adapter expansion.

Docs & diagrams

• AGENT_PROTOCOL_QUICKSTART.md – hands-on walkthrough for both client and server roles.
• AGENT_TO_AGENT_GUIDE.md – explains how users, A2A clients, and remote agents collaborate.
• docs/ARCHITECTURE.md – updated architecture diagram for the Agent Protocol–only build.
• docs/ADDING_PROVIDERS.md – how to add adapters back in using BaseAgent + AgentFactory.
• docs/SCHEMA_REGISTRY.md – describe the schema manifest format and how to plug custom schemas into the SDK.

Running tests

pip install -r requirements-dev.txt
pytest -v

The suite now focuses on:

• tests/test_agent_protocol_adapter.py
• tests/test_server.py
• tests/test_models.py
• tests/test_factory.py
• tests/test_config.py

GitHub Actions (.github/workflows/test.yml) runs these tests on every push and pull request, and .github/workflows/agent_protocol_e2e.yml spins up the sample FastAPI server to exercise the adapter end-to-end.

Extending later

Even though the repository now ships only the Agent Protocol adapter, the core scaffolding (base classes, registry, config helpers, packaging) stays intact. Follow docs/ADDING_PROVIDERS.md whenever you want to add another adapter, register new providers with AgentFactory, and reintroduce their tests/docs incrementally.

License

MIT – see LICENSE.