`server.py`

99 lines
"""Minimal A2A (Agent2Agent) demo server — a2a-sdk 1.1.2, current proto API.

An "Echo Agent": receives a text message and returns it upper-cased. It publishes
a standard A2A agent card at /.well-known/agent-card.json and serves the JSON-RPC
A2A endpoint at / . Proves the SDK server stack (AgentExecutor + DefaultRequestHandler
+ InMemoryTaskStore + Starlette route builders).

Run: . .venv/bin/activate && python server.py   (listens on 127.0.0.1:41241)
"""
from __future__ import annotations

import uuid

import uvicorn
from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.routes import create_agent_card_routes, create_jsonrpc_routes
from a2a.server.tasks import InMemoryTaskStore
from a2a.types import (
    AgentCapabilities,
    AgentCard,
    AgentInterface,
    AgentSkill,
    Message,
    Part,
    Role,
)
from starlette.applications import Starlette

HOST = "127.0.0.1"
PORT = 41241


class EchoAgentExecutor(AgentExecutor):
    """The agent's brain: read the user's text, reply with it upper-cased."""

    async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
        # get_user_input() concatenates the text parts of the incoming message.
        try:
            user_text = context.get_user_input() or ""
        except Exception:
            user_text = ""
        reply = Message(
            message_id=str(uuid.uuid4()),
            role=Role.ROLE_AGENT,
            parts=[Part(text=f"echo: {user_text.upper()}")],
        )
        await event_queue.enqueue_event(reply)

    async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
        raise NotImplementedError("Echo agent does not support cancellation.")


def build_agent_card() -> AgentCard:
    return AgentCard(
        name="Echo Agent",
        description="Minimal A2A demo agent — returns your text upper-cased.",
        version="0.1.0",
        supported_interfaces=[
            AgentInterface(
                url=f"http://{HOST}:{PORT}/",
                protocol_binding="JSONRPC",
                protocol_version="1.0",
            )
        ],
        capabilities=AgentCapabilities(streaming=False, push_notifications=False),
        default_input_modes=["text/plain"],
        default_output_modes=["text/plain"],
        skills=[
            AgentSkill(
                id="echo",
                name="Echo",
                description="Uppercase-echo the input text.",
                tags=["demo", "echo"],
                examples=["hello world"],
            )
        ],
    )


def build_app() -> Starlette:
    card = build_agent_card()
    handler = DefaultRequestHandler(
        agent_executor=EchoAgentExecutor(),
        task_store=InMemoryTaskStore(),
        agent_card=card,
    )
    routes = [
        *create_agent_card_routes(card),          # GET /.well-known/agent-card.json
        *create_jsonrpc_routes(handler, "/"),      # POST /  (A2A JSON-RPC)
    ]
    return Starlette(routes=routes)


if __name__ == "__main__":
    print(f"A2A Echo Agent on http://{HOST}:{PORT}  (card: /.well-known/agent-card.json)")
    uvicorn.run(build_app(), host=HOST, port=PORT, log_level="warning")