spakky-a2a

spakky-a2a는 spakky-agent를 A2A (Agent2Agent) protocol server와 원격 teammate delegation으로 노출하는 adapter plugin입니다. Spakky @Agent를 A2A server로 공개하고, agent spec/tool catalog/teammates에서 AgentCard를 유도하며, 공식 a2a-sdk client 위에 core IAgentDelegate port를 구현합니다.

설치

pip install spakky-a2a

실행 가능한 agent에는 별도 IAgentModel provider가 필요합니다. Durable run 또는 HITL resume을 사용하면 spakky-sqlalchemy[agent] 같은 provider가 공급하는 spakky-agent persistence repository도 필요합니다.

설정

A2AConfig는 SPAKKY_A2A_ 접두사의 환경변수를 읽습니다.

Plugin 초기화는 A2AConfig, A2AAgentRegistry, A2AAgentServerSpec, A2A remote delegate Pod, 그리고 @Agent와 @A2ACompatible가 함께 붙은 class를 발견해 ASGI/gRPC host에 연결하는 post-processor를 등록합니다.

Agent 노출

@A2ACompatible는 @Agent와 같은 class에 쌓는 tag입니다. @Agent가 Pod를 등록하고, tag는 A2A transport metadata와 optional mount path를 기록합니다.

from spakky.agent import Agent, AgentExecutionSpec, IAgentModel
from spakky.plugins.a2a import A2ACompatible


@A2ACompatible(
    base_url="https://agents.example.com/a2a/planner",
    version="1.0.0",
    mount_path="/a2a/planner",
    rest_mount_path="/a2a-rest/planner",
    rest_base_url="https://agents.example.com/a2a-rest/planner",
    grpc_enabled=True,
    grpc_base_url="grpc://agents.example.com:443",
)
@Agent(spec=AgentExecutionSpec(name="planner", objective="Plan work"))
class PlannerAgent:
    def __init__(self, model: IAgentModel) -> None:
        self.model = model

애플리케이션이 Starlette/FastAPI host Pod를 제공하면 plugin post-processor가 bootstrap 중 mount_path에 A2A JSON-RPC + AgentCard app을 자동 mount합니다. rest_mount_path를 지정하면 HTTP+JSON REST + AgentCard app도 그 path에 별도로 mount합니다. mount_path를 생략하면 {default_mount_path_prefix}/{agent_name}을 사용합니다. version을 생략하면 A2AConfig.default_version을 사용합니다.

base_url은 AgentCard supported_interfaces[].url에 광고되는 public transport endpoint입니다. 클라이언트가 reverse proxy나 ASGI mount path를 통해 호출한다면 그 외부 path를 포함해야 하며, /.well-known/agent-card.json 자체는 포함하지 않습니다. 예를 들어 위 선언에서 card URL은 https://agents.example.com/a2a/planner/.well-known/agent-card.json이고, JSON-RPC operation endpoint는 https://agents.example.com/a2a/planner/입니다. base_url을 생략하면 framework가 A2AConfig.default_base_url.rstrip("/") + mount_path로 유도하므로 기본 설정에서는 http://localhost:8000/a2a/planner가 광고됩니다. REST도 rest_base_url을 생략하면 default_base_url + rest_mount_path를 광고합니다.

from starlette.applications import Starlette
from spakky.core.pod.annotations.pod import Pod


@Pod(name="asgi_host")
def asgi_host() -> Starlette:
    return Starlette()

gRPC 노출은 grpc_enabled=True인 entry를 spakky-grpc의 GrpcServerSpec에 자동 등록합니다. 따라서 애플리케이션은 spakky-a2a와 spakky-grpc plugin을 함께 로드하고 SPAKKY_GRPC_BIND_ADDRESSES를 설정하면 됩니다. spakky-grpc가 로드되지 않은 애플리케이션에서는 gRPC 선언은 no-op입니다.

애플리케이션 bootstrap 이후 A2AAgentServerSpec.build_app_for("planner"), build_rest_app_for("planner"), build_grpc_handler_for("planner")는 특수 host나 테스트에서 쓰는 lower-level escape hatch입니다. 일반 애플리케이션은 @A2ACompatible 선언과 host Pod만 사용합니다.

AgentCard 유도

AgentCardFactory는 다음 입력에서 card를 유도합니다.

• name/description: AgentExecutionSpec.name, objective, 또는 instructions
• streaming capability: streaming_exposure_mode; NO_STREAM_UNTIL_FINAL_GUARDED는 streaming capability 노출을 끕니다.
• tool: synthetic teammate delegation tool을 제외한 native @agent_tool descriptor
• delegation skill: 선언된 AgentTeammate entry

실행별 입력

Executor는 A2A inbound message의 task id를 RunAgentInput.state_id로, context id를 conversation_id로 사용합니다. Message data part에 modelSelection 또는 model_selection을 넣으면 RunAgentInput.model_selection으로 전달되고, mcp와 metadata object는 RunAgentInput.metadata로 전달됩니다.

{
  "modelSelection": {"provider": "openrouter", "model": "anthropic/claude-sonnet-4.5"},
  "mcp": {"servers": ["github"]},
  "metadata": {"tenant": "acme"}
}

Approval resume은 data part의 approval_id, decision을 APPROVAL_DECISION signal로 변환합니다.

원격 Teammate 위임

A2AAgentDelegate는 AgentExecutionSpec.teammates entry가 원격 AgentCard URL을 가리키는 teammate를 위해 core IAgentDelegate port를 구현합니다. Plugin 초기화가 A2AAgentDelegate를 Pod로 등록하고 IAgentDelegate에 바인딩하므로 parent agent는 IAgentDelegate 또는 A2AAgentDelegate를 생성자 주입으로 받을 수 있습니다. Core agent runner는 각 teammate를 teammate.<schema_token(name)>.delegate라는 model-callable delegation tool로 노출합니다. schema_token은 teammate name의 앞뒤 공백을 제거한 뒤 [a-zA-Z0-9_]가 아닌 연속 문자를 단일 _로 치환하고, 앞뒤 _를 제거한 다음 소문자화한 값입니다. 이 결과가 비면 agent definition 단계에서 거부됩니다. Local teammate Pod는 in-process로 실행하고, remote teammate는 공식 a2a-sdk client를 사용합니다.

from spakky.agent import Agent, AgentExecutionSpec, AgentTeammate
from spakky.plugins.a2a import A2AAgentDelegate


@Agent(
    spec=AgentExecutionSpec(
        name="orchestrator",
        teammates=(
            AgentTeammate(
                name="researcher",
                card_url="https://agents.example.com/.well-known/agent-card.json",
            ),
        ),
    )
)
class Orchestrator:
    def __init__(self, delegate: A2AAgentDelegate) -> None:
        self.delegate = delegate

원격 delegation은 SDK client로 message/send를 보내고 remote task stream을 추적한 뒤, child task/message/artifact update를 parent run id를 보존한 Spakky protocol-neutral event stream으로 되돌립니다.

REST HTTP+JSON Transport

SDK route 이름은 JSON-RPC method 문자열과 다릅니다.

REST request/response body는 A2A SDK protobuf JSON encoding을 사용합니다. 예를 들어 user message는 {"message":{"role":"ROLE_USER","messageId":"m1","parts":[{"text":"hi"}]}} 형태로 보냅니다.

HITL와 Auth Interrupt

SpakkyAgentExecutor는 core AgentRunner.run_events() stream을 소비합니다. Approval/auth pause는 successful terminal RunFinishedEvent가 아니라 protocol-neutral RunPausedEvent로 들어옵니다. A2A projector는 reason=approval_required를 TASK_STATE_INPUT_REQUIRED로 매핑하고 approval id와 allowed decisions를 data part에 포함합니다. reason=auth_required는 TASK_STATE_AUTH_REQUIRED로 매핑하므로, run stream이 끝난 뒤 durable state.reason을 다시 조회하지 않아도 auth-required 상태를 표현할 수 있습니다.

Approval resume은 approval_id와 decision을 가진 inbound A2A data part로 전달됩니다. Executor는 APPROVAL_DECISION signal을 append하고 같은 task id로 RunAgentInput(resume=True)를 다시 실행합니다.

Task Store

Server transport는 synchronous IA2ATaskRepository port 위에 async a2a-sdk TaskStore를 얹는 bridge인 SpakkyA2ATaskStore를 사용합니다. Builder 인자로 repository를 주지 않고 container에도 repository Pod가 없으면 plugin은 InMemoryA2ATaskRepository를 사용합니다.

개발 검증

패키지 단위 검증은 해당 패키지 디렉토리에서 실행합니다.

uv run ruff format .
uv run ruff check .
uv run pyrefly check
uv run pytest

pytest는 각 패키지 pyproject.toml의 coverage 설정을 사용합니다.

라이선스

MIT License