A2A

tulip.a2a implements the Agent-to-Agent protocol: one agent calls another over HTTP, across process and organisation boundaries, without either side importing the other's code.

Two generations of the wire format ship side by side. The A2AV1* types are the v1 spec — use these for anything new. The unprefixed types are the earlier shape, kept because deployed peers still speak it; A2AServer accepts both.

For the concepts, start with the A2A protocol and the walkthrough notebook.

Client and server

A2AServer exposes an agent over the protocol; A2AClient calls one. Neither requires the other side to be built with Tulip.

A2AServer

A2AServer(agent: Any, name: str = 'Tulip Agent', description: str = '', skills: list[AgentSkill] | list[str] | None = None, url: str = '', provider: AgentProvider | None = None, version: str = '0.1.0', api_key: str | None = None, allow_unauthenticated: bool = False)

Bases: A2AV1ServerMixin

Expose a Tulip Agent as a spec-compliant A2A endpoint.

Parameters:

Name	Type	Description	Default
agent	Any	A Tulip Agent (or anything with run(prompt) -> AsyncIterator).	required
name	str	Display name for the Agent Card.	'Tulip Agent'
description	str	One-line description for the Agent Card.	''
skills	list[AgentSkill] | list[str] | None	List of :class:AgentSkill (preferred) or plain strings (legacy — auto-promoted to skills with id == name).	None
url	str	Public URL the agent is reachable at — set this for cross-process / cross-host deployments so the card's url field is correct. Defaults to a placeholder.	''
provider	AgentProvider | None	Optional :class:AgentProvider (e.g. Acme).	None
version	str	Agent semver — useful for capability negotiation.	'0.1.0'
api_key	str | None	Bearer token required on every route; if None, falls back to TULIP_A2A_API_KEY.	None
allow_unauthenticated	bool	Bind to non-loopback without a key. Use only behind an upstream proxy that terminates auth.	False

Example::

from tulip import Agent
from tulip.a2a import A2AServer
from tulip.a2a.spec import AgentSkill

server = A2AServer(
    agent=my_agent,
    name="Research Agent",
    description="Open-web research with citations.",
    skills=[
        AgentSkill(
            id="research",
            name="Research",
            description="Answer with cited sources.",
            tags=["search", "summarise"],
        ),
    ],
    url="https://research.example.com",
    api_key="secret",
)
server.run(port=8001)

Source code in .sdk/src/tulip/a2a/protocol.py

def __init__(
    self,
    agent: Any,
    name: str = "Tulip Agent",
    description: str = "",
    skills: list[AgentSkill] | list[str] | None = None,
    url: str = "",
    provider: AgentProvider | None = None,
    version: str = "0.1.0",
    api_key: str | None = None,
    allow_unauthenticated: bool = False,
) -> None:
    self._agent = agent
    self._name = name
    self._description = description or f"A2A-compatible {name}"
    self._skills = self._normalise_skills(skills)
    self._url = url
    self._provider = provider
    self._version = version
    self._api_key = api_key or os.environ.get("TULIP_A2A_API_KEY") or None
    self._allow_unauthenticated = allow_unauthenticated
    self._app: Any = None
    self._store = _TaskStore()

run

run(host: str = '127.0.0.1', port: int = 8001, **kwargs: Any) -> None

Run the A2A server.

Defaults to loopback binding. Non-loopback bindings require either api_key to be set or allow_unauthenticated=True.

Source code in .sdk/src/tulip/a2a/protocol.py

def run(self, host: str = "127.0.0.1", port: int = 8001, **kwargs: Any) -> None:
    """Run the A2A server.

    Defaults to loopback binding. Non-loopback bindings require
    either ``api_key`` to be set or ``allow_unauthenticated=True``.
    """
    if self._api_key is None and not self._allow_unauthenticated and not _is_loopback(host):
        msg = (
            f"Refusing to bind A2AServer to {host!r} without an API "
            "key. Set TULIP_A2A_API_KEY, pass api_key=... to "
            "A2AServer, or pass allow_unauthenticated=True if an "
            "upstream proxy terminates auth."
        )
        raise RuntimeError(msg)

    try:
        import uvicorn
    except ImportError as e:
        msg = "uvicorn required. Install with: pip install uvicorn"
        raise ImportError(msg) from e
    uvicorn.run(self.app, host=host, port=port, **kwargs)

A2AClient

A2AClient(url: str, api_key: str | None = None, timeout: float | Timeout | None = None, protocol_version: str | None = A2A_V1_PROTOCOL_VERSION)

Call a remote A2A agent from Tulip.

Spec-compliant methods:

• :meth:get_agent_card — fetches /.well-known/agent-card.json, falling back to the legacy /agent-card endpoint.
• :meth:send_message — JSON-RPC message/send; returns a :class:Task you can poll with :meth:get_task.
• :meth:send_message_streaming — JSON-RPC message/stream; yields events from the SSE stream.
• :meth:get_task, :meth:cancel_task — task lifecycle.

Plus the legacy convenience APIs preserved from the pre-spec implementation:

• :meth:invoke — flat string-in / string-out over /a2a/invoke.
• :meth:as_tool — wrap a remote agent as a Tulip @tool.

Source code in .sdk/src/tulip/a2a/protocol.py

def __init__(
    self,
    url: str,
    api_key: str | None = None,
    timeout: float | httpx.Timeout | None = None,
    protocol_version: str | None = A2A_V1_PROTOCOL_VERSION,
) -> None:
    self._url = url.rstrip("/")
    self._api_key = api_key
    self._timeout: Any = self.DEFAULT_TIMEOUT if timeout is None else timeout
    self._protocol_version = protocol_version

get_agent_card async

get_agent_card() -> AgentCard

Fetch the remote agent's capability card.

Tries the spec well-known URL first, falls back to the legacy /agent-card endpoint for older peers.

Source code in .sdk/src/tulip/a2a/protocol.py

async def get_agent_card(self) -> AgentCard:
    """Fetch the remote agent's capability card.

    Tries the spec well-known URL first, falls back to the legacy
    ``/agent-card`` endpoint for older peers.
    """
    import httpx

    async with httpx.AsyncClient() as client:
        for path in ("/.well-known/agent-card.json", "/agent-card"):
            try:
                resp = await client.get(f"{self._url}{path}", headers=self._headers())
                if resp.status_code == 200:
                    data = resp.json()
                    # Legacy peers serve flat string skills — promote.
                    if data.get("skills") and isinstance(data["skills"][0], str):
                        data["skills"] = [
                            {"id": s, "name": s, "description": s} for s in data["skills"]
                        ]
                    # Legacy peers also omit url/capabilities.
                    data.setdefault("url", self._url)
                    return AgentCard.model_validate(data)
            except httpx.HTTPError:
                continue
    msg = f"Could not fetch Agent Card from {self._url}"
    raise RuntimeError(msg)

send_message async

send_message(message: Message, *, timeout: float | Timeout | None = None) -> Task

Send a message via JSON-RPC message/send and return the Task.

Source code in .sdk/src/tulip/a2a/protocol.py

async def send_message(
    self,
    message: Message,
    *,
    timeout: float | httpx.Timeout | None = None,  # noqa: ASYNC109 — forwarded to httpx, not an asyncio timer
) -> Task:
    """Send a message via JSON-RPC ``message/send`` and return the Task."""
    if self._protocol_version == A2A_V1_PROTOCOL_VERSION:
        v1_message = legacy_message_to_v1(message)
        result = await self._rpc(
            "SendMessage",
            {"message": v1_message.model_dump(exclude_none=True)},
            timeout=timeout,
        )
        return Task.model_validate(task_result_to_legacy_payload(result))
    result = await self._rpc(
        "message/send",
        {"message": message.model_dump(exclude_none=True)},
        timeout=timeout,
    )
    return Task.model_validate(result)

send_message_streaming async

send_message_streaming(message: Message, *, timeout: float | Timeout | None = None) -> AsyncIterator[dict[str, Any]]

Send a message via JSON-RPC streaming and yield events.

Source code in .sdk/src/tulip/a2a/protocol.py

async def send_message_streaming(
    self,
    message: Message,
    *,
    timeout: float | httpx.Timeout | None = None,  # noqa: ASYNC109 — forwarded to httpx, not an asyncio timer
) -> AsyncIterator[dict[str, Any]]:
    """Send a message via JSON-RPC streaming and yield events."""
    import httpx

    if self._protocol_version == A2A_V1_PROTOCOL_VERSION:
        method = "SendStreamingMessage"
        message_payload = legacy_message_to_v1(message).model_dump(exclude_none=True)
    else:
        method = "message/stream"
        message_payload = message.model_dump(exclude_none=True)
    body = {
        "jsonrpc": "2.0",
        "id": uuid.uuid4().hex,
        "method": method,
        "params": {"message": message_payload},
    }
    headers = self._headers() | {"Accept": "text/event-stream"}
    effective_timeout = self._timeout if timeout is None else timeout
    async with (
        httpx.AsyncClient(timeout=effective_timeout) as client,
        client.stream("POST", f"{self._url}/", json=body, headers=headers) as resp,
    ):
        resp.raise_for_status()
        async for raw in resp.aiter_lines():
            if not raw or not raw.startswith("data: "):
                continue
            payload = raw[len("data: ") :]
            if payload.strip() == "[DONE]":
                break
            try:
                env = json.loads(payload)
            except json.JSONDecodeError:
                continue
            if "error" in env:
                yield env  # surface error envelope to caller
                break
            result = env.get("result", env)
            yield (
                v1_stream_response_to_legacy_payload(result)
                if method == "SendStreamingMessage"
                else result
            )

get_task async

get_task(task_id: str, history_length: int | None = None) -> Task

Fetch a task by id (JSON-RPC tasks/get).

Source code in .sdk/src/tulip/a2a/protocol.py

async def get_task(self, task_id: str, history_length: int | None = None) -> Task:
    """Fetch a task by id (JSON-RPC ``tasks/get``)."""
    params: dict[str, Any] = {"id": task_id}
    if history_length is not None:
        params["historyLength"] = history_length
    if self._protocol_version == A2A_V1_PROTOCOL_VERSION:
        result = await self._rpc("GetTask", params)
        return Task.model_validate(task_result_to_legacy_payload(result))
    result = await self._rpc("tasks/get", params)
    return Task.model_validate(result)

list_tasks async

list_tasks(*, context_id: str | None = None, status: TaskState | str | None = None, page_size: int | None = None, page_token: str | None = None, history_length: int | None = None, include_artifacts: bool | None = None) -> tuple[list[Task], str]

List known tasks.

Returns (tasks, next_page_token). The client maps v1.0 wire enum states back into SDK-shaped :class:Task objects.

Source code in .sdk/src/tulip/a2a/protocol.py

async def list_tasks(
    self,
    *,
    context_id: str | None = None,
    status: TaskState | str | None = None,
    page_size: int | None = None,
    page_token: str | None = None,
    history_length: int | None = None,
    include_artifacts: bool | None = None,
) -> tuple[list[Task], str]:
    """List known tasks.

    Returns ``(tasks, next_page_token)``. The client maps v1.0 wire
    enum states back into SDK-shaped :class:`Task` objects.
    """
    params: dict[str, Any] = {}
    if context_id is not None:
        params["contextId"] = context_id
    if status is not None:
        state = status.value if isinstance(status, TaskState) else str(status)
        if self._protocol_version == A2A_V1_PROTOCOL_VERSION and not state.startswith(
            "TASK_STATE_"
        ):
            state = "TASK_STATE_" + state.replace("-", "_").upper()
        params["status"] = state
    if page_size is not None:
        params["pageSize"] = page_size
    if page_token is not None:
        params["pageToken"] = page_token
    if history_length is not None:
        params["historyLength"] = history_length
    if include_artifacts is not None:
        params["includeArtifacts"] = include_artifacts

    if self._protocol_version == A2A_V1_PROTOCOL_VERSION:
        result = await self._rpc("ListTasks", params)
        tasks = [
            Task.model_validate(task_result_to_legacy_payload(task))
            for task in result.get("tasks", [])
        ]
        return tasks, str(result.get("nextPageToken", ""))

    raise RuntimeError("list_tasks requires A2A v1.0 protocol_version")

cancel_task async

cancel_task(task_id: str) -> Task

Cancel a task (JSON-RPC tasks/cancel).

Source code in .sdk/src/tulip/a2a/protocol.py

async def cancel_task(self, task_id: str) -> Task:
    """Cancel a task (JSON-RPC ``tasks/cancel``)."""
    if self._protocol_version == A2A_V1_PROTOCOL_VERSION:
        result = await self._rpc("CancelTask", {"id": task_id})
        return Task.model_validate(task_result_to_legacy_payload(result))
    result = await self._rpc("tasks/cancel", {"id": task_id})
    return Task.model_validate(result)

invoke async

invoke(prompt: str, *, timeout: float | Timeout | None = None) -> str

Send a flat text prompt over the legacy /a2a/invoke.

Useful when you control both ends of the wire and want a one-line round-trip; spec-compliant peers should prefer :meth:send_message so they can read the full :class:Task.

Source code in .sdk/src/tulip/a2a/protocol.py

async def invoke(
    self,
    prompt: str,
    *,
    timeout: float | httpx.Timeout | None = None,  # noqa: ASYNC109 — forwarded to httpx, not an asyncio timer
) -> str:
    """Send a flat text prompt over the legacy ``/a2a/invoke``.

    Useful when you control both ends of the wire and want a one-line
    round-trip; spec-compliant peers should prefer
    :meth:`send_message` so they can read the full :class:`Task`.
    """
    import httpx

    request = A2ARequest(messages=[A2AMessage(role="user", content=prompt)])
    async with httpx.AsyncClient(
        timeout=self._timeout if timeout is None else timeout
    ) as client:
        resp = await client.post(
            f"{self._url}/a2a/invoke",
            json=request.model_dump(),
            headers=self._auth_headers(),
        )
        resp.raise_for_status()
        response = A2AResponse.model_validate(resp.json())
    agent_msgs = [m for m in response.messages if m.role == "agent"]
    return agent_msgs[-1].content if agent_msgs else ""

as_tool

as_tool(name: str | None = None, description: str | None = None) -> Any

Wrap this remote agent as a Tulip @tool.

Source code in .sdk/src/tulip/a2a/protocol.py

def as_tool(self, name: str | None = None, description: str | None = None) -> Any:
    """Wrap this remote agent as a Tulip ``@tool``."""
    from tulip.tools.decorator import tool as tool_decorator

    client = self
    tool_name = name or "remote_agent"
    tool_desc = description or "Call a remote A2A agent"

    @tool_decorator(name=tool_name, description=tool_desc)
    def call_remote(prompt: str) -> str:
        """Send a request to a remote agent."""
        import asyncio

        return asyncio.run(client.invoke(prompt))

    return call_remote

Agent cards — discovery

An agent card is what a peer publishes about itself: who runs it, what it can do, and how to reach it. It is the only thing a caller needs before the first request.

AgentCard

Bases: BaseModel

Public Agent Card (spec §5.5).

Published at /.well-known/agent-card.json. The legacy /agent-card endpoint serves the same payload for backwards compatibility with peers that haven't picked up the well-known URL.

AgentSkill

Bases: BaseModel

A discrete capability the agent advertises in its card.

AgentCapabilities

Bases: BaseModel

Optional protocol-level capabilities the agent supports.

Per spec, these are declarations: a peer queries the agent card to know whether to attempt streaming or push-notification flows.

AgentProvider

Bases: BaseModel

The organization / publisher behind the agent.

AgentInterface

Bases: BaseModel

A concrete protocol binding exposed by an A2A v1.0 agent.

v1 protocol

The current wire format. A2A_V1_PROTOCOL_VERSION is the version string sent on the wire and the one a peer negotiates against.

A2A_V1_PROTOCOL_VERSION module-attribute

A2A_V1_PROTOCOL_VERSION = '1.0'

Sending a message

A2AV1SendMessageRequest

Bases: BaseModel

A2A v1.0 SendMessage request params.

A2AV1SendMessageResponse

Bases: BaseModel

A2A v1.0 SendMessageResponse oneof.

A2AV1SendMessageConfiguration

Bases: BaseModel

A2A v1.0 SendMessage configuration.

A2AV1Message

Bases: BaseModel

A2A v1.0 Message.

A2AV1Part

Bases: BaseModel

A2A v1.0 Part oneof.

Exactly one of text, data, raw or url should be set by callers. The model is intentionally permissive enough to round-trip extension fields that are outside Tulip's plain-text default.

A2AV1Role

Bases: StrEnum

A2A v1.0 message role enum names.

Tasks

A request that is not answered immediately becomes a task the caller polls or streams. A2AV1TaskState is the state machine; the update events are what a streaming caller receives.

A2AV1Task

Bases: BaseModel

A2A v1.0 Task.

A2AV1TaskState

Bases: StrEnum

A2A v1.0 task state enum names.

A2AV1TaskStatus

Bases: BaseModel

A2A v1.0 TaskStatus.

A2AV1GetTaskRequest

Bases: BaseModel

A2A v1.0 GetTask request params.

A2AV1CancelTaskRequest

Bases: BaseModel

A2A v1.0 CancelTask request params.

A2AV1ListTasksRequest

Bases: BaseModel

A2A v1.0 ListTasks request params.

A2AV1ListTasksResponse

Bases: BaseModel

A2A v1.0 ListTasks response.

Streaming and artifacts

A2AV1StreamResponse

Bases: BaseModel

A2A v1.0 StreamResponse oneof.

A2AV1TaskStatusUpdateEvent

Bases: BaseModel

A2A v1.0 TaskStatusUpdateEvent.

v1.0 stream completion is inferred from task state and stream closure; the pre-v1 final field is intentionally absent.

A2AV1TaskArtifactUpdateEvent

Bases: BaseModel

A2A v1.0 TaskArtifactUpdateEvent.

A2AV1Artifact

Bases: BaseModel

A2A v1.0 Artifact.

Messages and parts

A message is a list of parts. Part is the discriminated union — a part is text, a file, or structured data, and the kind field decides which.

Message

Bases: BaseModel

A user/agent message with one or more typed parts (spec §6.4).

Part module-attribute

Part = Annotated[TextPart | FilePart | DataPart, Field(discriminator='kind')]

TextPart

Bases: BaseModel

A plain-text message part.

DataPart

Bases: BaseModel

A structured-data (e.g. JSON) message part.

FilePart

Bases: BaseModel

A file message part — either inline bytes or a URI reference.

FileWithBytes

Bases: BaseModel

A file referenced by inline base64 bytes.

FileWithUri

Bases: BaseModel

A file referenced by URI.

Artifact

Bases: BaseModel

A typed result attached to a Task (spec §6.7).

Tasks

Task

Bases: BaseModel

A unit of work tracked through the lifecycle (spec §6.1).

TaskState

Bases: StrEnum

Task lifecycle states (spec §6.3).

TaskStatus

Bases: BaseModel

Status block on a Task (spec §6.2).

TaskIdParams

Bases: BaseModel

Identifier-only params (cancel, push-config get/list/delete).

TaskQueryParams

Bases: BaseModel

Parameters for tasks/get (spec §7.3).

TaskStatusUpdateEvent

Bases: BaseModel

Status transition event in the SSE stream (spec §7.2.2).

TaskArtifactUpdateEvent

Bases: BaseModel

Artifact-attach event in the SSE stream (spec §7.2.3).

MessageSendParams

Bases: BaseModel

Parameters for message/send and message/stream (spec §7.1).

MessageSendConfiguration

Bases: BaseModel

Optional per-call configuration (spec §7.1.4).

Push notifications

For work long enough that polling is the wrong shape: the peer calls you back when the task changes.

PushNotificationConfig

Bases: BaseModel

Webhook config attached to a Task for async updates.

TaskPushNotificationConfig

Bases: BaseModel

Bundle: which task + the webhook config (spec §7.5).

PushNotificationAuthenticationInfo

Bases: BaseModel

Auth shape for push-notification webhook delivery.

JSON-RPC envelope

The transport shapes. You rarely construct these directly — A2AClient and A2AServer do — but an error response is worth being able to read.

JsonRpcRequest

Bases: BaseModel

A JSON-RPC 2.0 request envelope.

The id MAY be omitted for notifications, but A2A's request methods always require a response so callers should always send one.

JsonRpcSuccessResponse

Bases: BaseModel

JSON-RPC 2.0 successful response.

JsonRpcErrorResponse

Bases: BaseModel

JSON-RPC 2.0 error response.

JsonRpcError

Bases: BaseModel

JSON-RPC 2.0 error object.

Legacy shapes

The pre-v1 request/response types. A2AServer still accepts them so deployed peers keep working; do not build anything new on them.

A2AMessage

Bases: BaseModel

Legacy flat message — preserved so peers + tests that still call /a2a/invoke keep working. Spec-aware peers should use :class:tulip.a2a.spec.Message.

A2ARequest

Bases: BaseModel

Legacy request envelope for POST /a2a/invoke.

A2AResponse

Bases: BaseModel

Legacy response envelope from POST /a2a/invoke.