A2A Protocol (ADR-038)

External agent interoperability via Agent-to-Agent Protocol.


Overview

Component Purpose
A2A Server Receive tasks from external agents
A2A Client Delegate tasks to external agents
Agent Card Capability discovery

Protocol Relationship

1
2
3
MCP:  Agent ↔ Tool    (how agents use capabilities)
A2A:  Agent ↔ Agent   (how agents collaborate)
SK:   Internal        (orchestration)

Agent Card

Served at /.well-known/agent.json:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
{
  "name": "sea-forge-agent",
  "version": "1.0.0",
  "description": "SEA-Forge™ semantic engineering agent",
  "url": "https://sea.example.com",
  "capabilities": {
    "streaming": true,
    "pushNotifications": true
  },
  "skills": [
    {
      "id": "semantic-analysis",
      "name": "Semantic Code Analysis",
      "description": "Analyze code semantics"
    },
    {
      "id": "artifact-generation",
      "name": "Artifact Generation",
      "description": "Generate diagrams and docs"
    }
  ],
  "authentication": {
    "schemes": ["bearer"]
  }
}

Task Lifecycle

State Description
submitted Task received
working Agent processing
input-required Waiting for input
completed Task finished
failed Task failed
canceled Task canceled

Sending Task (A2A Client)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
from a2a import A2AClient

client = A2AClient("https://external-agent.example.com")

task = await client.send_task({
    "skill": "code-review",
    "input": {
        "code": "function example() { ... }"
    }
})

# Stream results
async for update in client.stream_task(task.id):
    print(update)

Receiving Task (A2A Server)

1
2
3
4
5
6
7
8
9
10
11
from a2a import A2AServer

server = A2AServer()

@server.skill("semantic-analysis")
async def analyze(input: dict) -> dict:
    # Analyze code
    result = await semantic_kernel.invoke("analyze", input)
    return {"analysis": result}

server.run(port=8080)

Governance Integration

All A2A tasks pass through SDS-031:

1
2
3
4
5
6
7
@kernel.filter(FilterTypes.FUNCTION_INVOCATION)
async def hitl_filter(context, next):
    if context.function.metadata.get("requires_approval"):
        approval = await request_hitl_approval(context)
        if not approval.granted:
            raise HITLRejected(approval.reason)
    return await next(context)

Last Updated: January 2026