콘텐츠로 이동

spakky-a2a

spakky-a2a는 @Agent를 A2A AgentCard와 task transport로 노출하고, 원격 A2A teammate를 spakky-agent delegation stream으로 합류시키는 어댑터입니다.

서버 경로는 공식 a2a-sdk request handler와 task store를 사용하며, 실행은 IAgentRunnerFactory가 여는 runner의 AgentEvent stream을 A2A task/message/artifact update로 투영합니다. @A2ACompatible @Agent는 registry에 등록되고 ASGI host Pod가 있으면 JSON-RPC/REST endpoint가 자동 mount됩니다. spakky-grpc의 GrpcServerSpec가 있으면 gRPC handler도 선언형으로 등록됩니다.

Public API

A2A (Agent2Agent) protocol server plugin for the Spakky framework.

Exposes a spakky @Agent as an A2A protocol server: an AgentCard is derived from the agent's spec, tools, and teammates, and JSON-RPC/HTTP plus SSE routes are mounted from the official a2a-sdk. Marker, config, and plugin identifier are re-exported; transport types live under a2a-sdk.

SPAKKY_A2A_CONFIG_ENV_PREFIX = 'SPAKKY_A2A_' module-attribute

Environment prefix for A2A plugin settings.

A2AAgentServer = A2ACompatible module-attribute

Deprecated alias for :class:A2ACompatible.

PLUGIN_NAME = Plugin(name='spakky-a2a') module-attribute

Plugin identifier for the A2A integration.

A2AConfig()

Bases: BaseSettings

Configuration for the A2A protocol server integration.

Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/config.py 
40
41
def __init__(self) -> None:
    super().__init__()

default_base_url = DEFAULT_A2A_BASE_URL class-attribute instance-attribute

Public host URL combined with the mount path when a marker omits base_url.

default_version = DEFAULT_A2A_VERSION class-attribute instance-attribute

Semantic version advertised on a derived AgentCard.

default_mount_path_prefix = DEFAULT_A2A_MOUNT_PATH_PREFIX class-attribute instance-attribute

URL prefix under which discovered A2A agents are mounted.

A2ARemoteAgentClient(*, httpx_client=None, config=None)

Small wrapper around the official a2a-sdk client and types.

Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/client.py 
44
45
46
47
48
49
50
51
52
def __init__(
    self,
    *,
    httpx_client: httpx.AsyncClient | None = None,
    config: ClientConfig | None = None,
) -> None:
    self._httpx_client = httpx_client
    self._config = config or ClientConfig(httpx_client=httpx_client)
    self._factory = ClientFactory(self._config)

resolve_card(card_url) async

Fetch a remote AgentCard with the SDK resolver.

Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/client.py 
54
55
56
57
58
59
60
61
async def resolve_card(self, card_url: str) -> AgentCard:
    """Fetch a remote AgentCard with the SDK resolver."""
    parts = urlsplit(card_url)
    base_url = f"{parts.scheme}://{parts.netloc}"
    path = parts.path or DEFAULT_AGENT_CARD_PATH
    async with self._http_client() as client:
        resolver = A2ACardResolver(client, base_url=base_url)
        return await resolver.get_agent_card(path)

send_message(card_url, message) async

Send a message and collect the SDK response stream.

Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/client.py 
63
64
65
66
67
68
69
async def send_message(
    self,
    card_url: str,
    message: RemoteA2AMessage,
) -> tuple[StreamResponse, ...]:
    """Send a message and collect the SDK response stream."""
    return tuple([event async for event in self.stream_message(card_url, message)])

stream_message(card_url, message) async

Send a message and yield remote task/message updates as they arrive.

Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/client.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
async def stream_message(
    self,
    card_url: str,
    message: RemoteA2AMessage,
) -> AsyncGenerator[StreamResponse, None]:
    """Send a message and yield remote task/message updates as they arrive."""
    card = await self.resolve_card(card_url)
    client = self._factory.create(card)
    request = SendMessageRequest(
        message=Message(
            role=Role.ROLE_USER,
            message_id=message.message_id,
            task_id=message.task_id or "",
            context_id=message.context_id or "",
            parts=[Part(text=message.text)],
        ),
        configuration=SendMessageConfiguration(return_immediately=False),
    )
    try:
        async for event in client.send_message(request):
            yield event
    finally:
        await client.close()

get_task(card_url, task_id) async

Fetch a remote A2A task by id using the SDK client.

Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/client.py 
 95
 96
 97
 98
 99
100
101
102
async def get_task(self, card_url: str, task_id: str) -> Task:
    """Fetch a remote A2A task by id using the SDK client."""
    card = await self.resolve_card(card_url)
    client = self._factory.create(card)
    try:
        return await client.get_task(GetTaskRequest(id=task_id))
    finally:
        await client.close()

RemoteA2AMessage(text, task_id=None, context_id=None, message_id=(lambda: f'message-{uuid4()}')()) dataclass

Message envelope sent to a remote A2A teammate.

A2AAgentDelegate(client=A2ARemoteAgentClient(), mapper=A2AStreamEventMapper()) dataclass

Bases: IAgentDelegate

Delegate remote teammate calls through the official A2A client.

delegate(packet) async

Execute a remote delegation packet and yield its terminal result.

Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/delegation.py 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
async def delegate(
    self,
    packet: DelegationPacket,
) -> AsyncGenerator[AgentYield[DelegationResult], None]:
    """Execute a remote delegation packet and yield its terminal result."""
    tool_result = await self.delegate_tool_result(packet)
    yield AgentYield(
        kind=AgentYieldKind.FINAL,
        payload=DelegationResult(
            id=f"{packet.id}:result",
            packet_id=packet.id,
            target=packet.target,
            summary=tool_result.summary,
            output=tool_result.output,
            metadata=tool_result.metadata,
        ),
    )

delegate_tool_result(packet) async

Call a remote A2A teammate and return model result plus child events.

Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/delegation.py 
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
async def delegate_tool_result(
    self,
    packet: DelegationPacket,
) -> DelegationToolResult:
    """Call a remote A2A teammate and return model result plus child events."""
    card_url = _card_url(packet.target)
    events: list[AgentEvent] = []
    final_task: Task | None = None
    final_status: TaskStatusUpdateEvent | None = None
    async for response in self.client.stream_message(
        card_url,
        RemoteA2AMessage(
            text=_instruction(packet),
            context_id=_conversation_id(packet),
        ),
    ):
        events.extend(self.mapper.map(response, packet))
        if response.WhichOneof("payload") == "task":
            final_task = response.task
        if response.WhichOneof("payload") == "status_update":
            final_status = response.status_update
    output = _task_output(final_task, final_status)
    return DelegationToolResult(
        summary=_summary(packet, output),
        output=output,
        events=tuple(events),
        metadata={"packet_id": packet.id, "card_url": card_url},
    )

A2AStreamEventMapper

Map remote A2A SDK stream responses into neutral child events.

map(response, packet)

Project one SDK stream response onto neutral delegated events.

Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/delegation.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
def map(
    self,
    response: StreamResponse,
    packet: DelegationPacket,
) -> tuple[AgentEvent, ...]:
    """Project one SDK stream response onto neutral delegated events."""
    payload = response.WhichOneof("payload")
    if payload == "task":
        return self._task_events(response.task, packet)
    if payload == "message":
        return self._message_events(response.message, packet)
    if payload == "status_update":
        return self._status_events(response.status_update, packet)
    if payload == "artifact_update":
        return self._artifact_events(response.artifact_update, packet)
    return ()

A2AEndpointConflictError

Bases: AbstractSpakkyA2AError

Raised when multiple A2A agents claim the same ASGI mount path.

A2ARunResolutionError(field)

Bases: AbstractSpakkyA2AError

Raised when an inbound A2A request carries invalid run configuration.

Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/error.py 
79
80
81
def __init__(self, field: str) -> None:
    super().__init__()
    self.field = field

A2ACompatible(base_url=None, version=None, mount_path=None, rest_mount_path=None, rest_base_url=None, grpc_enabled=False, grpc_base_url=None) dataclass

Bases: Tag

Marks an @Agent class to be served through an A2A protocol endpoint.

Attributes:

Name Type Description
base_url str | None

Public transport endpoint advertised on the derived AgentCard.
version str | None

Semantic version advertised on the derived AgentCard.
mount_path str | None

Starlette/FastAPI mount path for automatic ASGI exposure.

base_url = None class-attribute instance-attribute

Public endpoint advertised on the derived AgentCard interface.

version = None class-attribute instance-attribute

Semantic version advertised on the derived AgentCard.

mount_path = None class-attribute instance-attribute

ASGI host mount path. None uses A2AConfig.default_mount_path_prefix.

rest_mount_path = None class-attribute instance-attribute

Optional ASGI mount path for the HTTP+JSON REST transport.

rest_base_url = None class-attribute instance-attribute

Public REST transport endpoint. None derives from default_base_url + path.

grpc_enabled = False class-attribute instance-attribute

Whether to register the official A2A gRPC handler when spakky-grpc is active.

grpc_base_url = None class-attribute instance-attribute

Public gRPC transport endpoint advertised by the gRPC handler.

설정

A2A plugin configuration.

SPAKKY_A2A_CONFIG_ENV_PREFIX = 'SPAKKY_A2A_' module-attribute

Environment prefix for A2A plugin settings.

DEFAULT_A2A_BASE_URL = 'http://localhost:8000' module-attribute

Fallback public host URL used to derive AgentCard transport endpoints.

DEFAULT_A2A_VERSION = '1.0.0' module-attribute

Fallback semantic version advertised on a derived AgentCard.

DEFAULT_A2A_MOUNT_PATH_PREFIX = '/a2a' module-attribute

Default Starlette/FastAPI mount prefix for discovered A2A agents.

A2AConfig()

Bases: BaseSettings

Configuration for the A2A protocol server integration.

Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/config.py 
40
41
def __init__(self) -> None:
    super().__init__()

default_base_url = DEFAULT_A2A_BASE_URL class-attribute instance-attribute

Public host URL combined with the mount path when a marker omits base_url.

default_version = DEFAULT_A2A_VERSION class-attribute instance-attribute

Semantic version advertised on a derived AgentCard.

default_mount_path_prefix = DEFAULT_A2A_MOUNT_PATH_PREFIX class-attribute instance-attribute

URL prefix under which discovered A2A agents are mounted.

AgentCard

AgentCard derivation from an @Agent declaration.

Maps a spakky @Agent spec, its discovered tool catalog, and its declared teammates onto an a2a-sdk AgentCard. The a2a-sdk 1.x AgentCard is a protobuf message (a2a_pb2) whose transport endpoint is expressed as an AgentInterface entry rather than a flat url field, so the base URL is advertised through supported_interfaces.

JSON_CONTENT_TYPE = 'application/json' module-attribute

Tool skills advertise JSON-shaped input and output payloads.

TEXT_CONTENT_TYPE = 'text/plain' module-attribute

The card's default conversational input and output content type.

TEAMMATE_DELEGATION_TAG = 'delegation' module-attribute

Tag attached to a skill derived from a declared teammate.

AgentCardFactory

Builds an a2a-sdk AgentCard from an @Agent Pod declaration.

build(agent, base_url, version, protocol=TransportProtocol.JSONRPC)

Derive an AgentCard from an @Agent spec, tools, and teammates.

Parameters:

Name Type Description Default
agent Agent

The @Agent Pod metadata carrying spec and tool catalog.

 required
base_url str

Transport endpoint advertised on the card interface.

 required
version str

Semantic version advertised on the card.

 required
protocol TransportProtocol

A2A transport protocol advertised for base_url.

 JSONRPC

Returns:

Type Description
AgentCard

A protobuf AgentCard ready to publish on the well-known route.
Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/card/derivation.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def build(
    self,
    agent: Agent,
    base_url: str,
    version: str,
    protocol: TransportProtocol = TransportProtocol.JSONRPC,
) -> AgentCard:
    """Derive an AgentCard from an @Agent spec, tools, and teammates.

    Args:
        agent: The @Agent Pod metadata carrying spec and tool catalog.
        base_url: Transport endpoint advertised on the card interface.
        version: Semantic version advertised on the card.
        protocol: A2A transport protocol advertised for ``base_url``.

    Returns:
        A protobuf ``AgentCard`` ready to publish on the well-known route.
    """
    spec = agent.spec
    name = spec.name or agent.target.__name__
    description = spec.objective or spec.instructions or name
    # A guarded final-only profile suppresses incremental streaming exposure.
    streaming = (
        spec.streaming_exposure_mode
        is not StreamingExposureMode.NO_STREAM_UNTIL_FINAL_GUARDED
    )
    tool_skills = [
        self._tool_skill(descriptor)
        for descriptor in agent.tool_catalog.descriptors
        if not self._is_teammate_delegation_tool(descriptor)
    ]
    teammate_skills = [
        self._teammate_skill(teammate) for teammate in spec.teammates
    ]
    return AgentCard(
        name=name,
        description=description,
        version=version,
        supported_interfaces=[
            AgentInterface(
                url=base_url,
                protocol_binding=protocol.value,
            )
        ],
        capabilities=AgentCapabilities(
            streaming=streaming,
            push_notifications=False,
        ),
        default_input_modes=[TEXT_CONTENT_TYPE],
        default_output_modes=[TEXT_CONTENT_TYPE],
        skills=[*tool_skills, *teammate_skills],
    )

서버 등록

@A2ACompatible marker for exposing an @Agent over the A2A protocol.

Unlike @GrpcController (which subclasses Controller and registers its own Pod), this marker subclasses :class:~spakky.core.pod.annotations.tag.Tag: the @Agent decorator already registers the Pod, so this tag only records the A2A transport metadata and stacks above @Agent on the same class.

A2ACompatible(base_url=None, version=None, mount_path=None, rest_mount_path=None, rest_base_url=None, grpc_enabled=False, grpc_base_url=None) dataclass

Bases: Tag

Marks an @Agent class to be served through an A2A protocol endpoint.

Attributes:

Name Type Description
base_url str | None

Public transport endpoint advertised on the derived AgentCard.
version str | None

Semantic version advertised on the derived AgentCard.
mount_path str | None

Starlette/FastAPI mount path for automatic ASGI exposure.

base_url = None class-attribute instance-attribute

Public endpoint advertised on the derived AgentCard interface.

version = None class-attribute instance-attribute

Semantic version advertised on the derived AgentCard.

mount_path = None class-attribute instance-attribute

ASGI host mount path. None uses A2AConfig.default_mount_path_prefix.

rest_mount_path = None class-attribute instance-attribute

Optional ASGI mount path for the HTTP+JSON REST transport.

rest_base_url = None class-attribute instance-attribute

Public REST transport endpoint. None derives from default_base_url + path.

grpc_enabled = False class-attribute instance-attribute

Whether to register the official A2A gRPC handler when spakky-grpc is active.

grpc_base_url = None class-attribute instance-attribute

Public gRPC transport endpoint advertised by the gRPC handler.

Registry of @Agent instances exposed as A2A servers.

A2AAgentServerEntry(agent_name, instance, agent_type, metadata) dataclass

A discovered @Agent instance paired with its A2A transport metadata.

A2AAgentRegistry()

Holds the @Agent instances discovered as A2A servers, keyed by name.

Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/server/registry.py 
28
29
def __init__(self) -> None:
    self._entries = {}

register(instance, agent_type, metadata)

Register an @Agent instance under its resolved agent name.

Parameters:

Name Type Description Default
instance object

The @Agent Pod instance to serve.

 required
agent_type type[object]

The original @Agent type, unwrapped from AOP proxies.

 required
metadata A2ACompatible

The A2A transport metadata declared on the agent class.

 required
Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/server/registry.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
def register(
    self,
    instance: object,
    agent_type: type[object],
    metadata: A2ACompatible,
) -> None:
    """Register an @Agent instance under its resolved agent name.

    Args:
        instance: The @Agent Pod instance to serve.
        agent_type: The original @Agent type, unwrapped from AOP proxies.
        metadata: The A2A transport metadata declared on the agent class.
    """
    agent_name = self._agent_name(agent_type)
    self._entries[agent_name] = A2AAgentServerEntry(
        agent_name=agent_name,
        instance=instance,
        agent_type=agent_type,
        metadata=metadata,
    )

get(agent_name)

Return the registered entry for an agent name.

Raises:

Type Description
A2AAgentServerNotRegisteredError

No entry exists for the name.
Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/server/registry.py 
52
53
54
55
56
57
58
59
60
61
def get(self, agent_name: str) -> A2AAgentServerEntry:
    """Return the registered entry for an agent name.

    Raises:
        A2AAgentServerNotRegisteredError: No entry exists for the name.
    """
    entry = self._entries.get(agent_name)
    if entry is None:
        raise A2AAgentServerNotRegisteredError(agent_name)
    return entry

list_entries()

Return registered A2A agent entries in stable name order.

Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/server/registry.py 
63
64
65
def list_entries(self) -> tuple[A2AAgentServerEntry, ...]:
    """Return registered A2A agent entries in stable name order."""
    return tuple(self._entries[name] for name in sorted(self._entries))

Assembly of a mountable A2A ASGI application for one @Agent instance.

The a2a-sdk 1.x server is assembled from route factories rather than a single application class: the agent-card route plus the JSON-RPC routes (with v0.3 compatibility enabling the message/send / tasks/get method names) are mounted on a Starlette app the host application can further mount.

A2AAgentServerSpec

Bases: IContainerAware

Container-aware factory that builds A2A apps for registered agents.

build_app_for(agent_name)

Build a mountable A2A app for a registered agent name.

Resolves the registry entry, then an optional task repository Pod from the container, falling back to an in-memory store when none is registered.

Parameters:

Name Type Description Default
agent_name str

The registered agent name to serve.

 required

Returns:

Type Description
Starlette

A Starlette application for the named agent.
Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/server/builder.py 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def build_app_for(self, agent_name: str) -> Starlette:
    """Build a mountable A2A app for a registered agent name.

    Resolves the registry entry, then an optional task repository Pod from the
    container, falling back to an in-memory store when none is registered.

    Args:
        agent_name: The registered agent name to serve.

    Returns:
        A Starlette application for the named agent.
    """
    entry = self._container.get(A2AAgentRegistry).get(agent_name)
    # Repository Pod is optional; absent one, persistence stays in-process.
    repository = self._container.get_or_none(IA2ATaskRepository)
    runner_factory = self._container.get(IAgentRunnerFactory)
    return build_a2a_app(
        entry.instance,
        base_url=self._base_url(entry),
        version=self._version(entry),
        repository=repository,
        agent_type=entry.agent_type,
        runner_factory=runner_factory,
    )

build_rest_app_for(agent_name)

Build a mountable A2A REST app for a registered agent name.

Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/server/builder.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
def build_rest_app_for(self, agent_name: str) -> Starlette:
    """Build a mountable A2A REST app for a registered agent name."""
    entry = self._container.get(A2AAgentRegistry).get(agent_name)
    repository = self._container.get_or_none(IA2ATaskRepository)
    runner_factory = self._container.get(IAgentRunnerFactory)
    return build_a2a_rest_app(
        entry.instance,
        base_url=self._rest_base_url(entry),
        version=self._version(entry),
        repository=repository,
        agent_type=entry.agent_type,
        runner_factory=runner_factory,
    )

build_grpc_handler_for(agent_name)

Build an A2A gRPC handler for a registered agent name.

Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/server/builder.py 
122
123
124
125
126
127
128
129
130
131
132
133
134
def build_grpc_handler_for(self, agent_name: str) -> A2AGrpcHandler:
    """Build an A2A gRPC handler for a registered agent name."""
    entry = self._container.get(A2AAgentRegistry).get(agent_name)
    repository = self._container.get_or_none(IA2ATaskRepository)
    runner_factory = self._container.get(IAgentRunnerFactory)
    return build_a2a_grpc_handler(
        entry.instance,
        base_url=self._grpc_base_url(entry),
        version=self._version(entry),
        repository=repository,
        agent_type=entry.agent_type,
        runner_factory=runner_factory,
    )

mount_path_for(agent_name)

Return the ASGI mount path for a registered agent.

Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/server/builder.py 
136
137
138
139
140
141
142
def mount_path_for(self, agent_name: str) -> str:
    """Return the ASGI mount path for a registered agent."""
    entry = self._container.get(A2AAgentRegistry).get(agent_name)
    if entry.metadata.mount_path is not None:
        return entry.metadata.mount_path
    prefix = self._config().default_mount_path_prefix.rstrip("/")
    return f"{prefix}/{agent_name}"

rest_mount_path_for(agent_name)

Return the optional REST ASGI mount path for a registered agent.

Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/server/builder.py 
144
145
146
147
def rest_mount_path_for(self, agent_name: str) -> str | None:
    """Return the optional REST ASGI mount path for a registered agent."""
    entry = self._container.get(A2AAgentRegistry).get(agent_name)
    return entry.metadata.rest_mount_path

build_a2a_app(agent_instance, *, base_url, version, repository=None, agent_type=None, runner_factory=None)

Assemble a mountable A2A ASGI application for an @Agent instance.

Parameters:

Name Type Description Default
agent_instance object

The @Agent Pod instance to serve.

 required
base_url str

Transport endpoint advertised on the derived AgentCard.

 required
version str

Semantic version advertised on the derived AgentCard.

 required
repository IA2ATaskRepository | None

Task persistence port; an in-memory store is used when None.

 None
agent_type type[object] | None

Original @Agent class, supplied when the instance is proxied.

 None

Returns:

Type Description
Starlette

A Starlette application exposing the agent-card and JSON-RPC routes.
Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/server/builder.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
def build_a2a_app(
    agent_instance: object,
    *,
    base_url: str,
    version: str,
    repository: IA2ATaskRepository | None = None,
    agent_type: type[object] | None = None,
    runner_factory: IAgentRunnerFactory | None = None,
) -> Starlette:
    """Assemble a mountable A2A ASGI application for an @Agent instance.

    Args:
        agent_instance: The @Agent Pod instance to serve.
        base_url: Transport endpoint advertised on the derived AgentCard.
        version: Semantic version advertised on the derived AgentCard.
        repository: Task persistence port; an in-memory store is used when None.
        agent_type: Original @Agent class, supplied when the instance is proxied.

    Returns:
        A Starlette application exposing the agent-card and JSON-RPC routes.
    """
    card = AgentCardFactory().build(
        Agent.get(agent_type or type(agent_instance)),
        base_url,
        version,
    )
    handler = build_a2a_request_handler(
        agent_instance,
        base_url=base_url,
        version=version,
        repository=repository,
        agent_type=agent_type,
        card=card,
        runner_factory=runner_factory,
    )
    routes = [
        *create_agent_card_routes(card),
        *create_jsonrpc_routes(handler, DEFAULT_RPC_URL, enable_v0_3_compat=True),
    ]
    return Starlette(routes=routes)

Shared A2A request-handler assembly for all server transports.

build_a2a_request_handler(agent_instance, *, base_url, version, repository=None, agent_type=None, protocol=TransportProtocol.JSONRPC, card=None, runner_factory=None)

Build the official SDK request handler shared by A2A transports.

Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/server/request_handler.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
def build_a2a_request_handler(
    agent_instance: object,
    *,
    base_url: str,
    version: str,
    repository: IA2ATaskRepository | None = None,
    agent_type: type | None = None,
    protocol: TransportProtocol = TransportProtocol.JSONRPC,
    card: AgentCard | None = None,
    runner_factory: IAgentRunnerFactory | None = None,
) -> DefaultRequestHandler:
    """Build the official SDK request handler shared by A2A transports."""
    agent_card = card or AgentCardFactory().build(
        Agent.get(agent_type or type(agent_instance)),
        base_url,
        version,
        protocol=protocol,
    )
    store = SpakkyA2ATaskStore(repository or InMemoryA2ATaskRepository())
    executor = SpakkyAgentExecutor(
        agent_instance,
        AgentEventProjector(),
        runner_factory=runner_factory,
    )
    return DefaultRequestHandler(
        agent_executor=executor,
        task_store=store,
        agent_card=agent_card,
    )

Executor

a2a-sdk AgentExecutor bound to a spakky @Agent instance.

Drives the framework-owned :class:~spakky.agent.runner.AgentRunner for one A2A request over the neutral run_events() stream and projects each AgentEvent onto A2A task events. The A2A task id seeds the agent's durable state_id so a human-approval pause resumes on the same run when the caller sends the next message with the same task id and an approval-decision data part.

The runner emits approval/auth interruptions as first-class RunPausedEvent items, so the A2A projector maps those directly to input-required or auth-required. This executor only reconciles ordinary RUN_FINISHED success/failure after the stream drains.

APPROVAL_ID_PART_KEY = 'approval_id' module-attribute

Inbound data-part key carrying the approval request id to resume.

APPROVAL_DECISION_PART_KEY = 'decision' module-attribute

Inbound data-part key carrying the chosen approval decision value.

MODEL_SELECTION_PART_KEY = 'modelSelection' module-attribute

Inbound data-part key carrying a run-scoped provider/model selector.

MODEL_SELECTION_SNAKE_PART_KEY = 'model_selection' module-attribute

Snake-case model selection key accepted for non-JavaScript A2A clients.

RUN_METADATA_PART_KEY = 'metadata' module-attribute

Inbound data-part key carrying extra core RunAgentInput metadata.

MCP_PART_KEY = 'mcp' module-attribute

Inbound data-part key carrying runtime MCP server selectors.

RUN_FAILED_FALLBACK_MESSAGE = 'run failed' module-attribute

Status message used when a failed run carries no error message.

SpakkyAgentExecutor(agent, projector, runner_factory=None)

Bases: AgentExecutor

Bridges A2A request execution onto the spakky agent event stream.

Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/executor/adapter.py 
76
77
78
79
80
81
82
83
84
def __init__(
    self,
    agent: object,
    projector: AgentEventProjector,
    runner_factory: IAgentRunnerFactory | None = None,
) -> None:
    self._agent = agent
    self._projector = projector
    self._runner_factory = runner_factory or AgentRunnerFactory()

Projection of neutral AgentEvent items onto A2A task events.

The runner's run_events() emits the protocol-neutral AgentEvent taxonomy (ADR-0013 §3); this projector reproduces each event one-to-one as an a2a-sdk task update. A2A 1.x parts are protobuf messages: text is Part(text=...) and structured data is Part(data=<google.protobuf.Value>), built from a JSON-compatible value via ParseDict.

RUN_FINISHED is not applied as a terminal transition here: the executor owns the single complete/failed terminal update after draining the stream. Neutral RUN_PAUSED events are different: they are already non-terminal protocol interrupts, so this projector maps them directly to A2A input-required or auth-required task states.

RunOutcome(error, paused=False) dataclass

Terminal or interrupt result of one run, reconciled by the executor.

error is None for a successful run and carries the runner's terminal failure payload otherwise. paused is true after a RUN_PAUSED event has already applied the non-terminal task transition.

AgentEventProjector

Projects neutral AgentEvent items onto a2a-sdk task-event updates.

project(event, updater) async

Publish A2A events for one agent event via the task updater.

Parameters:

Name Type Description Default
event AgentEvent

The neutral event emitted by the agent runner.

 required
updater TaskUpdater

The a2a-sdk updater bound to the running task.

 required

Returns:

Type Description
RunOutcome | None

The run's terminal outcome for a RUN_FINISHED event, else None.

Raises:

Type Description
UnsupportedAgentEventError

The event kind has no A2A projection.
Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/executor/event_mapping.py 
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
async def project(
    self,
    event: AgentEvent,
    updater: TaskUpdater,
) -> RunOutcome | None:
    """Publish A2A events for one agent event via the task updater.

    Args:
        event: The neutral event emitted by the agent runner.
        updater: The a2a-sdk updater bound to the running task.

    Returns:
        The run's terminal outcome for a ``RUN_FINISHED`` event, else None.

    Raises:
        UnsupportedAgentEventError: The event kind has no A2A projection.
    """
    match event.kind:
        case AgentEventKind.RUN_STARTED:
            await updater.start_work()
        case AgentEventKind.RUN_FINISHED:
            return RunOutcome(error=_as(event, RunFinishedEvent).error)
        case AgentEventKind.RUN_PAUSED:
            await self._project_run_paused(_as(event, RunPausedEvent), updater)
            return RunOutcome(error=None, paused=True)
        case AgentEventKind.STEP_STARTED:
            await self._project_step(
                _as(event, StepStartedEvent).step_name, updater
            )
        case AgentEventKind.STEP_FINISHED:
            await self._project_step(
                _as(event, StepFinishedEvent).step_name, updater
            )
        case AgentEventKind.MESSAGE_DELTA:
            await self._project_message_delta(event, updater)
        case AgentEventKind.REASONING_DELTA:
            await self._project_reasoning_delta(event, updater)
        case AgentEventKind.TOOL_CALL_START:
            await self._project_tool_call_start(event, updater)
        case AgentEventKind.TOOL_CALL_ARGS_DELTA:
            await self._project_tool_call_args(event, updater)
        case AgentEventKind.TOOL_CALL_END:
            await self._project_tool_call_end(event, updater)
        case AgentEventKind.TOOL_CALL_RESULT:
            await self._project_tool_call_result(event, updater)
        case AgentEventKind.ARTIFACT:
            await self._project_artifact(event, updater)
        case AgentEventKind.STATE_SNAPSHOT:
            await self._project_state_snapshot(event, updater)
        case AgentEventKind.STATE_DELTA:
            await self._project_state_delta(event, updater)
        case _:  # pragma: no cover - exhaustive AgentEventKind StrEnum
            raise UnsupportedAgentEventError(str(event.kind))
    return None

Transports

HTTP+JSON REST transport bindings for the official A2A routes.

build_a2a_rest_app(agent_instance, *, base_url, version, repository=None, agent_type=None, path_prefix='', runner_factory=None)

Build a mountable A2A HTTP+JSON REST app for one @Agent instance.

Parameters:

Name Type Description Default
agent_instance object

The @Agent Pod instance to serve.

 required
base_url str

Transport endpoint advertised on the derived AgentCard.

 required
version str

Semantic version advertised on the derived AgentCard.

 required
repository IA2ATaskRepository | None

Task persistence port; an in-memory store is used when None.

 None
agent_type type | None

Original @Agent class, supplied when the instance is proxied.

 None
path_prefix str

Optional URL prefix for the REST operation routes.

 ''

Returns:

Type Description
Starlette

A Starlette application exposing AgentCard plus HTTP+JSON A2A routes.
Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/rest_transport/builder.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
def build_a2a_rest_app(
    agent_instance: object,
    *,
    base_url: str,
    version: str,
    repository: IA2ATaskRepository | None = None,
    agent_type: type | None = None,
    path_prefix: str = "",
    runner_factory: IAgentRunnerFactory | None = None,
) -> Starlette:
    """Build a mountable A2A HTTP+JSON REST app for one @Agent instance.

    Args:
        agent_instance: The @Agent Pod instance to serve.
        base_url: Transport endpoint advertised on the derived AgentCard.
        version: Semantic version advertised on the derived AgentCard.
        repository: Task persistence port; an in-memory store is used when None.
        agent_type: Original @Agent class, supplied when the instance is proxied.
        path_prefix: Optional URL prefix for the REST operation routes.

    Returns:
        A Starlette application exposing AgentCard plus HTTP+JSON A2A routes.
    """
    card = AgentCardFactory().build(
        Agent.get(agent_type or type(agent_instance)),
        base_url,
        version,
        protocol=TransportProtocol.HTTP_JSON,
    )
    handler = build_a2a_request_handler(
        agent_instance,
        base_url=base_url,
        version=version,
        repository=repository,
        agent_type=agent_type,
        protocol=TransportProtocol.HTTP_JSON,
        card=card,
        runner_factory=runner_factory,
    )
    routes = [
        *create_agent_card_routes(card),
        *create_rest_routes(
            handler,
            enable_v0_3_compat=True,
            path_prefix=path_prefix,
        ),
    ]
    return Starlette(routes=routes)

Assembly helpers for the A2A HTTP+JSON REST transport.

build_a2a_rest_app(agent_instance, *, base_url, version, repository=None, agent_type=None, path_prefix='', runner_factory=None)

Build a mountable A2A HTTP+JSON REST app for one @Agent instance.

Parameters:

Name Type Description Default
agent_instance object

The @Agent Pod instance to serve.

 required
base_url str

Transport endpoint advertised on the derived AgentCard.

 required
version str

Semantic version advertised on the derived AgentCard.

 required
repository IA2ATaskRepository | None

Task persistence port; an in-memory store is used when None.

 None
agent_type type | None

Original @Agent class, supplied when the instance is proxied.

 None
path_prefix str

Optional URL prefix for the REST operation routes.

 ''

Returns:

Type Description
Starlette

A Starlette application exposing AgentCard plus HTTP+JSON A2A routes.
Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/rest_transport/builder.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
def build_a2a_rest_app(
    agent_instance: object,
    *,
    base_url: str,
    version: str,
    repository: IA2ATaskRepository | None = None,
    agent_type: type | None = None,
    path_prefix: str = "",
    runner_factory: IAgentRunnerFactory | None = None,
) -> Starlette:
    """Build a mountable A2A HTTP+JSON REST app for one @Agent instance.

    Args:
        agent_instance: The @Agent Pod instance to serve.
        base_url: Transport endpoint advertised on the derived AgentCard.
        version: Semantic version advertised on the derived AgentCard.
        repository: Task persistence port; an in-memory store is used when None.
        agent_type: Original @Agent class, supplied when the instance is proxied.
        path_prefix: Optional URL prefix for the REST operation routes.

    Returns:
        A Starlette application exposing AgentCard plus HTTP+JSON A2A routes.
    """
    card = AgentCardFactory().build(
        Agent.get(agent_type or type(agent_instance)),
        base_url,
        version,
        protocol=TransportProtocol.HTTP_JSON,
    )
    handler = build_a2a_request_handler(
        agent_instance,
        base_url=base_url,
        version=version,
        repository=repository,
        agent_type=agent_type,
        protocol=TransportProtocol.HTTP_JSON,
        card=card,
        runner_factory=runner_factory,
    )
    routes = [
        *create_agent_card_routes(card),
        *create_rest_routes(
            handler,
            enable_v0_3_compat=True,
            path_prefix=path_prefix,
        ),
    ]
    return Starlette(routes=routes)

gRPC transport bindings for the official A2A service descriptor.

A2A_GRPC_SERVICE = 'lf.a2a.v1.A2AService' module-attribute

Fully qualified official A2A gRPC service name from the a2a-sdk descriptor.

A2AGrpcHandler(handler)

Bases: GenericRpcHandler

Generic gRPC handler for the official A2A service methods.

Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/grpc_transport/handler.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def __init__(self, handler: DefaultRequestHandler) -> None:
    self._handler = handler
    self._handlers = {
        self._method("SendMessage"): grpc.unary_unary_rpc_method_handler(
            self._send_message,
            request_deserializer=_deserializer(SendMessageRequest),
            response_serializer=_serializer,
        ),
        self._method("SendStreamingMessage"): grpc.unary_stream_rpc_method_handler(
            self._send_streaming_message,
            request_deserializer=_deserializer(SendMessageRequest),
            response_serializer=_serializer,
        ),
        self._method("GetTask"): grpc.unary_unary_rpc_method_handler(
            self._get_task,
            request_deserializer=_deserializer(GetTaskRequest),
            response_serializer=_serializer,
        ),
        self._method("CancelTask"): grpc.unary_unary_rpc_method_handler(
            self._cancel_task,
            request_deserializer=_deserializer(CancelTaskRequest),
            response_serializer=_serializer,
        ),
    }

service(handler_call_details)

Return the method handler for an official A2A gRPC path.

Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/grpc_transport/handler.py 
70
71
72
73
74
75
76
@override
def service(
    self,
    handler_call_details: grpc.HandlerCallDetails,
) -> grpc.RpcMethodHandler | None:
    """Return the method handler for an official A2A gRPC path."""
    return self._handlers.get(handler_call_details.method)

build_a2a_grpc_handler(agent_instance, *, base_url, version, repository=None, agent_type=None, runner_factory=None)

Build a gRPC handler for one @Agent-backed A2A server.

Parameters:

Name Type Description Default
agent_instance object

The @Agent Pod instance to serve.

 required
base_url str

Transport endpoint advertised on the derived AgentCard.

 required
version str

Semantic version advertised on the derived AgentCard.

 required
repository IA2ATaskRepository | None

Task persistence port; an in-memory store is used when None.

 None
agent_type type | None

Original @Agent class, supplied when the instance is proxied.

 None

Returns:

Type Description
A2AGrpcHandler

A generic gRPC handler exposing the official A2A service methods.
Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/grpc_transport/builder.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def build_a2a_grpc_handler(
    agent_instance: object,
    *,
    base_url: str,
    version: str,
    repository: IA2ATaskRepository | None = None,
    agent_type: type | None = None,
    runner_factory: IAgentRunnerFactory | None = None,
) -> A2AGrpcHandler:
    """Build a gRPC handler for one @Agent-backed A2A server.

    Args:
        agent_instance: The @Agent Pod instance to serve.
        base_url: Transport endpoint advertised on the derived AgentCard.
        version: Semantic version advertised on the derived AgentCard.
        repository: Task persistence port; an in-memory store is used when None.
        agent_type: Original @Agent class, supplied when the instance is proxied.

    Returns:
        A generic gRPC handler exposing the official A2A service methods.
    """
    request_handler = build_a2a_request_handler(
        agent_instance,
        base_url=base_url,
        version=version,
        repository=repository,
        agent_type=agent_type,
        protocol=TransportProtocol.GRPC,
        runner_factory=runner_factory,
    )
    return A2AGrpcHandler(request_handler)

Assembly helpers for the A2A gRPC transport.

build_a2a_grpc_handler(agent_instance, *, base_url, version, repository=None, agent_type=None, runner_factory=None)

Build a gRPC handler for one @Agent-backed A2A server.

Parameters:

Name Type Description Default
agent_instance object

The @Agent Pod instance to serve.

 required
base_url str

Transport endpoint advertised on the derived AgentCard.

 required
version str

Semantic version advertised on the derived AgentCard.

 required
repository IA2ATaskRepository | None

Task persistence port; an in-memory store is used when None.

 None
agent_type type | None

Original @Agent class, supplied when the instance is proxied.

 None

Returns:

Type Description
A2AGrpcHandler

A generic gRPC handler exposing the official A2A service methods.
Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/grpc_transport/builder.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def build_a2a_grpc_handler(
    agent_instance: object,
    *,
    base_url: str,
    version: str,
    repository: IA2ATaskRepository | None = None,
    agent_type: type | None = None,
    runner_factory: IAgentRunnerFactory | None = None,
) -> A2AGrpcHandler:
    """Build a gRPC handler for one @Agent-backed A2A server.

    Args:
        agent_instance: The @Agent Pod instance to serve.
        base_url: Transport endpoint advertised on the derived AgentCard.
        version: Semantic version advertised on the derived AgentCard.
        repository: Task persistence port; an in-memory store is used when None.
        agent_type: Original @Agent class, supplied when the instance is proxied.

    Returns:
        A generic gRPC handler exposing the official A2A service methods.
    """
    request_handler = build_a2a_request_handler(
        agent_instance,
        base_url=base_url,
        version=version,
        repository=repository,
        agent_type=agent_type,
        protocol=TransportProtocol.GRPC,
        runner_factory=runner_factory,
    )
    return A2AGrpcHandler(request_handler)

Official A2A gRPC service handler backed by the existing A2A executor.

The a2a-sdk ships protobuf descriptors for lf.a2a.v1.A2AService. This handler binds those official method names to the same DefaultRequestHandler used by the JSON-RPC transport, so AgentCard derivation, task persistence, executor adaptation, and neutral agent-event projection remain shared.

A2A_GRPC_SERVICE = 'lf.a2a.v1.A2AService' module-attribute

Fully qualified official A2A gRPC service name from the a2a-sdk descriptor.

A2AGrpcHandler(handler)

Bases: GenericRpcHandler

Generic gRPC handler for the official A2A service methods.

Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/grpc_transport/handler.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def __init__(self, handler: DefaultRequestHandler) -> None:
    self._handler = handler
    self._handlers = {
        self._method("SendMessage"): grpc.unary_unary_rpc_method_handler(
            self._send_message,
            request_deserializer=_deserializer(SendMessageRequest),
            response_serializer=_serializer,
        ),
        self._method("SendStreamingMessage"): grpc.unary_stream_rpc_method_handler(
            self._send_streaming_message,
            request_deserializer=_deserializer(SendMessageRequest),
            response_serializer=_serializer,
        ),
        self._method("GetTask"): grpc.unary_unary_rpc_method_handler(
            self._get_task,
            request_deserializer=_deserializer(GetTaskRequest),
            response_serializer=_serializer,
        ),
        self._method("CancelTask"): grpc.unary_unary_rpc_method_handler(
            self._cancel_task,
            request_deserializer=_deserializer(CancelTaskRequest),
            response_serializer=_serializer,
        ),
    }

service(handler_call_details)

Return the method handler for an official A2A gRPC path.

Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/grpc_transport/handler.py 
70
71
72
73
74
75
76
@override
def service(
    self,
    handler_call_details: grpc.HandlerCallDetails,
) -> grpc.RpcMethodHandler | None:
    """Return the method handler for an official A2A gRPC path."""
    return self._handlers.get(handler_call_details.method)

Client와 Delegation

Official a2a-sdk client wrapper for remote teammate calls.

DEFAULT_AGENT_CARD_PATH = '/.well-known/agent-card.json' module-attribute

Default A2A well-known AgentCard route.

RemoteA2AMessage(text, task_id=None, context_id=None, message_id=(lambda: f'message-{uuid4()}')()) dataclass

Message envelope sent to a remote A2A teammate.

A2ARemoteAgentClient(*, httpx_client=None, config=None)

Small wrapper around the official a2a-sdk client and types.

Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/client.py 
44
45
46
47
48
49
50
51
52
def __init__(
    self,
    *,
    httpx_client: httpx.AsyncClient | None = None,
    config: ClientConfig | None = None,
) -> None:
    self._httpx_client = httpx_client
    self._config = config or ClientConfig(httpx_client=httpx_client)
    self._factory = ClientFactory(self._config)

resolve_card(card_url) async

Fetch a remote AgentCard with the SDK resolver.

Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/client.py 
54
55
56
57
58
59
60
61
async def resolve_card(self, card_url: str) -> AgentCard:
    """Fetch a remote AgentCard with the SDK resolver."""
    parts = urlsplit(card_url)
    base_url = f"{parts.scheme}://{parts.netloc}"
    path = parts.path or DEFAULT_AGENT_CARD_PATH
    async with self._http_client() as client:
        resolver = A2ACardResolver(client, base_url=base_url)
        return await resolver.get_agent_card(path)

send_message(card_url, message) async

Send a message and collect the SDK response stream.

Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/client.py 
63
64
65
66
67
68
69
async def send_message(
    self,
    card_url: str,
    message: RemoteA2AMessage,
) -> tuple[StreamResponse, ...]:
    """Send a message and collect the SDK response stream."""
    return tuple([event async for event in self.stream_message(card_url, message)])

stream_message(card_url, message) async

Send a message and yield remote task/message updates as they arrive.

Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/client.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
async def stream_message(
    self,
    card_url: str,
    message: RemoteA2AMessage,
) -> AsyncGenerator[StreamResponse, None]:
    """Send a message and yield remote task/message updates as they arrive."""
    card = await self.resolve_card(card_url)
    client = self._factory.create(card)
    request = SendMessageRequest(
        message=Message(
            role=Role.ROLE_USER,
            message_id=message.message_id,
            task_id=message.task_id or "",
            context_id=message.context_id or "",
            parts=[Part(text=message.text)],
        ),
        configuration=SendMessageConfiguration(return_immediately=False),
    )
    try:
        async for event in client.send_message(request):
            yield event
    finally:
        await client.close()

get_task(card_url, task_id) async

Fetch a remote A2A task by id using the SDK client.

Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/client.py 
 95
 96
 97
 98
 99
100
101
102
async def get_task(self, card_url: str, task_id: str) -> Task:
    """Fetch a remote A2A task by id using the SDK client."""
    card = await self.resolve_card(card_url)
    client = self._factory.create(card)
    try:
        return await client.get_task(GetTaskRequest(id=task_id))
    finally:
        await client.close()

A2A-backed teammate delegation for @Agent teammate specs.

A2AStreamEventMapper

Map remote A2A SDK stream responses into neutral child events.

map(response, packet)

Project one SDK stream response onto neutral delegated events.

Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/delegation.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
def map(
    self,
    response: StreamResponse,
    packet: DelegationPacket,
) -> tuple[AgentEvent, ...]:
    """Project one SDK stream response onto neutral delegated events."""
    payload = response.WhichOneof("payload")
    if payload == "task":
        return self._task_events(response.task, packet)
    if payload == "message":
        return self._message_events(response.message, packet)
    if payload == "status_update":
        return self._status_events(response.status_update, packet)
    if payload == "artifact_update":
        return self._artifact_events(response.artifact_update, packet)
    return ()

A2AAgentDelegate(client=A2ARemoteAgentClient(), mapper=A2AStreamEventMapper()) dataclass

Bases: IAgentDelegate

Delegate remote teammate calls through the official A2A client.

delegate(packet) async

Execute a remote delegation packet and yield its terminal result.

Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/delegation.py 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
async def delegate(
    self,
    packet: DelegationPacket,
) -> AsyncGenerator[AgentYield[DelegationResult], None]:
    """Execute a remote delegation packet and yield its terminal result."""
    tool_result = await self.delegate_tool_result(packet)
    yield AgentYield(
        kind=AgentYieldKind.FINAL,
        payload=DelegationResult(
            id=f"{packet.id}:result",
            packet_id=packet.id,
            target=packet.target,
            summary=tool_result.summary,
            output=tool_result.output,
            metadata=tool_result.metadata,
        ),
    )

delegate_tool_result(packet) async

Call a remote A2A teammate and return model result plus child events.

Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/delegation.py 
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
async def delegate_tool_result(
    self,
    packet: DelegationPacket,
) -> DelegationToolResult:
    """Call a remote A2A teammate and return model result plus child events."""
    card_url = _card_url(packet.target)
    events: list[AgentEvent] = []
    final_task: Task | None = None
    final_status: TaskStatusUpdateEvent | None = None
    async for response in self.client.stream_message(
        card_url,
        RemoteA2AMessage(
            text=_instruction(packet),
            context_id=_conversation_id(packet),
        ),
    ):
        events.extend(self.mapper.map(response, packet))
        if response.WhichOneof("payload") == "task":
            final_task = response.task
        if response.WhichOneof("payload") == "status_update":
            final_status = response.status_update
    output = _task_output(final_task, final_status)
    return DelegationToolResult(
        summary=_summary(packet, output),
        output=output,
        events=tuple(events),
        metadata={"packet_id": packet.id, "card_url": card_url},
    )

Task Store

Plugin-owned persistence port for A2A task state.

The a2a-sdk TaskStore ABC is async and threads a ServerCallContext through every call. This plugin owns a narrower synchronous repository port so that adapters (in-memory today, a database-backed implementation later) stay free of a2a-sdk server types; the async bridge lives in task_store.

IA2ATaskRepository

Bases: ABC

Synchronous persistence port for A2A Task snapshots.

get_or_none(task_id) abstractmethod

Return a persisted task by id, or None when absent.

Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/store/interfaces.py 
18
19
20
21
@abstractmethod
def get_or_none(self, task_id: str) -> Task | None:
    """Return a persisted task by id, or None when absent."""
    ...

save(task) abstractmethod

Persist or overwrite a task snapshot.

Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/store/interfaces.py 
23
24
25
26
@abstractmethod
def save(self, task: Task) -> None:
    """Persist or overwrite a task snapshot."""
    ...

delete(task_id) abstractmethod

Remove a task snapshot by id.

Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/store/interfaces.py 
28
29
30
31
@abstractmethod
def delete(self, task_id: str) -> None:
    """Remove a task snapshot by id."""
    ...

list_all() abstractmethod

Return every persisted task snapshot.

Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/store/interfaces.py 
33
34
35
36
@abstractmethod
def list_all(self) -> Sequence[Task]:
    """Return every persisted task snapshot."""
    ...

In-memory task repository plus the async a2a-sdk TaskStore bridge.

InMemoryA2ATaskRepository()

Bases: IA2ATaskRepository

Dictionary-backed synchronous A2A task repository.

Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/store/task_store.py 
18
19
def __init__(self) -> None:
    self._tasks = {}

SpakkyA2ATaskStore(repository)

Bases: TaskStore

Async a2a-sdk TaskStore delegating to a synchronous repository.

The a2a-sdk request handler awaits every store call, but the plugin's repository port is synchronous; each async method calls straight through to the in-process repository, which performs no I/O of its own.

Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/store/task_store.py 
48
49
def __init__(self, repository: IA2ATaskRepository) -> None:
    self._repository = repository

Plugin

Post-processor registering @A2ACompatible-marked @Agent Pods.

RegisterA2AAgentServersPostProcessor

Bases: IPostProcessor, IContainerAware

Registers Pods carrying both @Agent and @A2ACompatible in the registry.

post_process(pod)

Register pod when it is an @A2ACompatible-marked @Agent.

Pods missing either marker are returned unchanged.

Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/post_processors/register_agent_servers.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
@override
def post_process(self, pod: object) -> object:
    """Register *pod* when it is an @A2ACompatible-marked @Agent.

    Pods missing either marker are returned unchanged.
    """
    pod_type = self._unwrap_proxy_type(type(pod))
    if not (A2ACompatible.exists(pod_type) and Agent.exists(pod_type)):
        return pod
    registry = self._container.get(A2AAgentRegistry)
    registry.register(pod, pod_type, A2ACompatible.get(pod_type))
    logger.info("Registered A2A agent server from %s", pod_type.__qualname__)
    return pod

Post-processor that mounts discovered A2A agents on ASGI host Pods.

MountA2AASGIPostProcessor()

Bases: IPostProcessor, IContainerAware, IApplicationContextAware

Mount registered A2A agent apps on Starlette-compatible host Pods.

Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/post_processors/mount_asgi.py 
40
41
42
def __init__(self) -> None:
    self._claimed_paths = {}
    self._mounted = set()

post_process(pod)

Mount registered A2A servers when a host or marked agent appears.

Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/post_processors/mount_asgi.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
@override
def post_process(self, pod: object) -> object:
    """Mount registered A2A servers when a host or marked agent appears."""
    if isinstance(pod, Starlette):
        self._mount_registered_agents(pod)
        return pod
    agent_type = self._unwrap_proxy_type(type(pod))
    if not (A2ACompatible.exists(agent_type) and Agent.exists(agent_type)):
        return pod
    entry = self._container.get(A2AAgentRegistry).get(self._agent_name(agent_type))
    for app in self._asgi_hosts():
        self._mount_entry(app, entry)
    logger.info("Mounted A2A agent server from %s", agent_type.__qualname__)
    return pod

Post-processor registering A2A gRPC handlers with spakky-grpc.

GRPC_SERVER_SPEC_MODULE = 'spakky.plugins.grpc.server_spec' module-attribute

Module path used to identify spakky-grpc's GrpcServerSpec without importing it.

GRPC_SERVER_SPEC_NAME = 'GrpcServerSpec' module-attribute

Class name used to identify spakky-grpc's GrpcServerSpec without importing it.

RegisterA2AGRPCPostProcessor()

Bases: IPostProcessor, IContainerAware

Register @A2ACompatible gRPC handlers when spakky-grpc is active.

Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/post_processors/register_grpc.py 
41
42
43
def __init__(self) -> None:
    self._registered = set()
    self._grpc_specs = []

post_process(pod)

Register enabled A2A gRPC handlers when a spec or agent appears.

Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/post_processors/register_grpc.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
@override
def post_process(self, pod: object) -> object:
    """Register enabled A2A gRPC handlers when a spec or agent appears."""
    if self._is_grpc_spec(pod):
        grpc_spec = cast(_GrpcServerSpecLike, pod)
        self._grpc_specs.append(grpc_spec)
        for entry in self._container.get(A2AAgentRegistry).list_entries():
            self._register_entry(grpc_spec, entry.agent_name)
        return pod
    if self._is_marked_agent(pod):
        agent_type = self._unwrap_proxy_type(type(pod))
        agent_name = self._agent_name(agent_type)
        for grpc_spec in self._grpc_specs:
            self._register_entry(grpc_spec, agent_name)
    return pod

Plugin initialization for the A2A protocol server integration.

Registers the plugin configuration, the agent-server registry, the container-aware app-builder spec, and the post-processor that discovers @A2ACompatible-marked @Agent Pods. This function is called automatically during plugin loading.

initialize(app)

Initialize the A2A plugin.

Parameters:

Name Type Description Default
app SpakkyApplication

The Spakky application instance.

 required
Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/main.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
def initialize(app: SpakkyApplication) -> None:
    """Initialize the A2A plugin.

    Args:
        app: The Spakky application instance.
    """
    if not app.container.contains(IAgentRunnerFactory):
        app.add(AgentRunnerFactory)
    app.add(A2AConfig)
    app.add(A2AAgentDelegate)
    app.container.bind_to_type(IAgentDelegate, A2AAgentDelegate)
    app.add(A2AAgentRegistry)
    app.add(A2AAgentServerSpec)
    app.add(RegisterA2AAgentServersPostProcessor)
    app.add(MountA2AASGIPostProcessor)
    app.add(RegisterA2AGRPCPostProcessor)

에러

A2A plugin error hierarchy.

Provides the base error class plus concrete errors raised while deriving an AgentCard, projecting neutral agent events onto A2A task events, and resolving a registered A2A agent server.

AbstractSpakkyA2AError

Bases: AbstractSpakkyFrameworkError, ABC

Base exception for all Spakky A2A plugin errors.

A2AAgentServerNotRegisteredError(agent_name)

Bases: AbstractSpakkyA2AError

Raised when no A2A agent server is registered for a requested name.

Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/error.py 
23
24
25
def __init__(self, agent_name: str) -> None:
    super().__init__()
    self.agent_name = agent_name

A2AAgentCardDerivationError(agent_name)

Bases: AbstractSpakkyA2AError

Raised when an AgentCard cannot be derived from an @Agent declaration.

Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/error.py 
33
34
35
def __init__(self, agent_name: str) -> None:
    super().__init__()
    self.agent_name = agent_name

A2AEndpointConflictError

Bases: AbstractSpakkyA2AError

Raised when multiple A2A agents claim the same ASGI mount path.

UnsupportedAgentEventError(kind)

Bases: AbstractSpakkyA2AError

Raised when an AgentEvent kind has no A2A task-event projection.

Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/error.py 
49
50
51
def __init__(self, kind: str) -> None:
    super().__init__()
    self.kind = kind

UnsupportedFinalOutputError(output_type)

Bases: AbstractSpakkyA2AError

Raised when a final agent output cannot be projected to an A2A part.

Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/error.py 
59
60
61
def __init__(self, output_type: type[object]) -> None:
    super().__init__()
    self.output_type = output_type

InvalidApprovalDecisionError(decision)

Bases: AbstractSpakkyA2AError

Raised when an inbound approval-decision part carries an unknown decision.

Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/error.py 
69
70
71
def __init__(self, decision: str) -> None:
    super().__init__()
    self.decision = decision

A2ARunResolutionError(field)

Bases: AbstractSpakkyA2AError

Raised when an inbound A2A request carries invalid run configuration.

Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/error.py 
79
80
81
def __init__(self, field: str) -> None:
    super().__init__()
    self.field = field

UnsupportedA2AGrpcResultError(result_type)

Bases: AbstractSpakkyA2AError

Raised when a unary A2A gRPC method returns an unsupported result.

Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/error.py 
89
90
91
def __init__(self, result_type: type[object]) -> None:
    super().__init__()
    self.result_type = result_type

UnsupportedA2AGrpcEventError(event_type)

Bases: AbstractSpakkyA2AError

Raised when a streaming A2A gRPC event cannot be wrapped.

Source code in plugins/spakky-a2a/src/spakky/plugins/a2a/error.py 
 99
100
101
def __init__(self, event_type: type[object]) -> None:
    super().__init__()
    self.event_type = event_type