Skip to content

A2A Protocol

Agent-to-Agent protocol support for inter-agent communication.

CognitiaA2AAdapter

CognitiaA2AAdapter

Adapts a Cognitia Agent to the A2A protocol.

Parameters:

Name Type Description Default
agent Any

A Cognitia Agent instance.

 required
name str

Agent name for the AgentCard.

 'Cognitia Agent'
description str | None

Agent description for the AgentCard.

 None
url str

Base URL where this agent's A2A server will be hosted.

 required
skills list[AgentSkill] | None

List of AgentSkill objects to advertise.

 None
version str

A2A card version string.

 '1.0'

agent_card property 

agent_card: AgentCard

Generate the AgentCard for discovery.

handle_task async 

handle_task(task: Task) -> Task

Process a task synchronously (non-streaming).

  1. Extracts user message text from task.messages
  2. Runs agent.query()
  3. Returns updated task with agent response

handle_task_streaming async 

handle_task_streaming(task: Task) -> AsyncIterator[TaskStatusUpdateEvent]

Process a task with SSE streaming.

Yields TaskStatusUpdateEvent as the task progresses.

get_task 

get_task(task_id: str) -> Task | None

Retrieve a task by ID.

cancel_task async 

cancel_task(task_id: str) -> Task | None

Cancel a task.

A2AServer

A2AServer

HTTP server for the A2A protocol.

Parameters:

Name Type Description Default
adapter Any

CognitiaA2AAdapter instance wrapping a Cognitia Agent.

 required
host str

Bind host. Default: "0.0.0.0".

 '127.0.0.1'
port int

Bind port. Default: 8000.

 8000

app property 

app: Any

Get or create the Starlette ASGI application.

serve async 

serve() -> None

Start the server with uvicorn.

Requires: pip install uvicorn

A2AClient

A2AClient

Client for communicating with remote A2A agents.

Parameters:

Name Type Description Default
url str

Base URL of the A2A server (e.g. "http://localhost:8000").

 required
timeout float

HTTP request timeout in seconds. Default: 30.

 30.0

discover async 

discover() -> AgentCard

Fetch the agent's AgentCard from /.well-known/agent.json.

Returns:

Type Description
AgentCard

The remote agent's capabilities and metadata.

send_task async 

send_task(message: str, *, task_id: str | None = None, session_id: str | None = None, metadata: dict[str, Any] | None = None) -> Task

Send a task to the remote agent (non-streaming).

Parameters:

Name Type Description Default
message str

User message text.

 required
task_id str | None

Optional task ID. Generated if not provided.

 None
session_id str | None

Optional session ID for conversation continuity.

 None
metadata dict[str, Any] | None

Optional metadata dict attached to the task.

 None

Returns:

Type Description
Task

The completed (or failed) task with agent response.

stream_task async 

stream_task(message: str, *, task_id: str | None = None, session_id: str | None = None) -> AsyncIterator[TaskStatusUpdateEvent]

Send a task with SSE streaming.

Parameters:

Name Type Description Default
message str

User message text.

 required
task_id str | None

Optional task ID.

 None
session_id str | None

Optional session ID.

 None

Yields:

Type Description
TaskStatusUpdateEvent

Status updates as the task progresses.

get_task async 

get_task(task_id: str) -> Task

Retrieve a task by ID.

Parameters:

Name Type Description Default
task_id str

The task ID to retrieve.

 required

Returns:

Type Description
Task

The task with current status.

cancel_task async 

cancel_task(task_id: str) -> Task

Cancel a task.

Parameters:

Name Type Description Default
task_id str

The task ID to cancel.

 required

Returns:

Type Description
Task

The task with updated (canceled) status.

Types

Task

Bases: BaseModel

A2A Task — unit of work between agents.

Lifecycle: submitted → working → completed/failed/canceled May transition to input-required if agent needs more info.

AgentCard

Bases: BaseModel

A2A Agent Card — discoverable at /.well-known/agent.json.

Describes an agent's identity, capabilities, and skills.

Message

Bases: BaseModel

A2A Message — a single turn in a conversation.

role: "user" (caller) or "agent" (callee) parts: list of TextPart, DataPart, or FilePart

TaskStatus

Bases: BaseModel

Current status of a task with optional message.

TaskState

Bases: str, Enum

A2A Task lifecycle states.