A2A API Reference¶

class JSONRPCDispatcher:
    """Routes JSON-RPC method calls to registered handler functions.

    Example::

        dispatcher = JSONRPCDispatcher()

        @dispatcher.register("SendMessage")
        def handle_send(params):
            ...
            return result_dict
    """

    def __init__(self) -> None:
        self._handlers: dict[str, MethodHandler] = {}

    def register(self, method: str) -> Callable[[MethodHandler], MethodHandler]:
        """Decorator to register a handler for *method*.

        Args:
            method: JSON-RPC method name (e.g. ``"SendMessage"``).

        Returns:
            The original handler function, unmodified.
        """

        def decorator(fn: MethodHandler) -> MethodHandler:
            self._handlers[method] = fn
            return fn

        return decorator

    def dispatch(
        self, request: JSONRPCRequest
    ) -> Union[JSONRPCResponse, Iterator[JSONRPCResponse]]:
        """Dispatch a parsed JSON-RPC request to the appropriate handler.

        Catches ``JSONRPCException`` (and subclasses) raised by handlers and
        converts them to error responses.

        Args:
            request: The parsed JSON-RPC request.

        Returns:
            A single ``JSONRPCResponse`` or, for streaming methods, a
            generator yielding ``JSONRPCResponse`` objects.
        """
        handler = self._handlers.get(request.method)
        if handler is None:
            return JSONRPCResponse(
                id=request.id,
                error=JSONRPCError(
                    code=METHOD_NOT_FOUND,
                    message=f"Method not found: {request.method}",
                ),
            )
        try:
            result = handler(request.params or {})
            if hasattr(result, "__next__"):
                return self._stream_wrap(request.id, result)
            return JSONRPCResponse.success(request.id, result)
        except JSONRPCException as exc:
            return JSONRPCResponse.from_error(request.id, exc)
        except Exception as exc:
            logger.exception("Unhandled error in handler %s", request.method)
            return JSONRPCResponse(
                id=request.id,
                error=JSONRPCError(code=INTERNAL_ERROR, message=str(exc)),
            )

    @staticmethod
    def _stream_wrap(request_id: Any, gen: Iterator[Any]) -> Iterator[JSONRPCResponse]:
        """Wrap a generator so each yielded value becomes a JSONRPCResponse."""
        try:
            for item in gen:
                yield JSONRPCResponse.success(request_id, item)
        except JSONRPCException as exc:
            yield JSONRPCResponse.from_error(request_id, exc)
        except Exception as exc:
            logger.exception("Unhandled error in streaming handler")
            yield JSONRPCResponse(
                id=request_id,
                error=JSONRPCError(code=INTERNAL_ERROR, message=str(exc)),
            )

@dataclass
class JSONRPCError:
    """A JSON-RPC 2.0 error object.

    Attributes:
        code: Numeric error code.
        message: Human-readable error message.
        data: Optional additional error data.
    """

    code: int = INTERNAL_ERROR
    message: str = "Internal error"
    data: Any = None

    def to_dict(self) -> dict[str, Any]:
        """Serialize to a dictionary."""
        d: dict[str, Any] = {"code": self.code, "message": self.message}
        if self.data is not None:
            d["data"] = self.data
        return d

    @classmethod
    def from_dict(cls, raw: dict[str, Any]) -> JSONRPCError:
        """Deserialize from a dictionary."""
        return cls(
            code=raw["code"],
            message=raw["message"],
            data=raw.get("data"),
        )

@dataclass
class JSONRPCRequest:
    """A JSON-RPC 2.0 request object.

    Attributes:
        method: The RPC method name.
        params: Method parameters.
        id: Request identifier (``None`` for notifications).
        jsonrpc: Protocol version (always ``"2.0"``).
    """

    method: str = ""
    params: dict[str, Any] | None = None
    id: Union[str, int, None] = None
    jsonrpc: str = JSONRPC_VERSION

    @property
    def is_notification(self) -> bool:
        """Whether this request is a notification (no ``id``)."""
        return self.id is None

    def to_dict(self) -> dict[str, Any]:
        """Serialize to a dictionary."""
        d: dict[str, Any] = {"jsonrpc": self.jsonrpc, "method": self.method}
        if self.id is not None:
            d["id"] = self.id
        if self.params is not None:
            d["params"] = self.params
        return d

    @classmethod
    def from_dict(cls, d: dict[str, Any]) -> JSONRPCRequest:
        """Deserialize from a dictionary."""
        return cls(
            method=d.get("method", ""),
            params=d.get("params"),
            id=d.get("id"),
            jsonrpc=d.get("jsonrpc", JSONRPC_VERSION),
        )

@dataclass
class JSONRPCResponse:
    """A JSON-RPC 2.0 response object.

    Exactly one of ``result`` or ``error`` should be set.

    Attributes:
        id: Matching request identifier.
        result: Successful result payload.
        error: Error details on failure.
        jsonrpc: Protocol version (always ``"2.0"``).
    """

    id: Union[str, int, None] = None
    result: Any = None
    error: JSONRPCError | None = None
    jsonrpc: str = JSONRPC_VERSION

    def to_dict(self) -> dict[str, Any]:
        """Serialize to a dictionary."""
        d: dict[str, Any] = {"jsonrpc": self.jsonrpc, "id": self.id}
        if self.error is not None:
            d["error"] = self.error.to_dict()
        else:
            d["result"] = self.result
        return d

    @classmethod
    def from_dict(cls, d: dict[str, Any]) -> JSONRPCResponse:
        """Deserialize from a dictionary."""
        error_raw = d.get("error")
        return cls(
            id=d.get("id"),
            result=d.get("result"),
            error=JSONRPCError.from_dict(error_raw) if error_raw else None,
            jsonrpc=d.get("jsonrpc", JSONRPC_VERSION),
        )

    @classmethod
    def success(cls, request_id: Any, result: Any) -> JSONRPCResponse:
        """Create a successful response."""
        return cls(id=request_id, result=result)

    @classmethod
    def from_error(cls, request_id: Any, error: Any) -> JSONRPCResponse:
        """Create an error response.

        Args:
            request_id: The id of the original request.
            error: A ``JSONRPCError`` dataclass, a ``JSONRPCException``
                (via its ``.error`` attribute), or any object with
                ``.code``, ``.rpc_message``, and ``.data`` attributes.
        """
        if isinstance(error, JSONRPCError):
            return cls(id=request_id, error=error)
        if isinstance(error, JSONRPCException):
            return cls(id=request_id, error=error.error)
        # Duck-typed protocol error (e.g. A2AError with code/rpc_message/data)
        return cls(
            id=request_id,
            error=JSONRPCError(
                code=getattr(error, "code", INTERNAL_ERROR),
                message=getattr(error, "rpc_message", str(error)),
                data=getattr(error, "data", None),
            ),
        )

class TaskState(str, enum.Enum):
    """Lifecycle states of a Task (mirrors ``TaskState`` proto enum)."""

    UNSPECIFIED = "TASK_STATE_UNSPECIFIED"
    SUBMITTED = "TASK_STATE_SUBMITTED"
    WORKING = "TASK_STATE_WORKING"
    COMPLETED = "TASK_STATE_COMPLETED"
    FAILED = "TASK_STATE_FAILED"
    CANCELED = "TASK_STATE_CANCELED"
    INPUT_REQUIRED = "TASK_STATE_INPUT_REQUIRED"
    REJECTED = "TASK_STATE_REJECTED"
    AUTH_REQUIRED = "TASK_STATE_AUTH_REQUIRED"

    def is_terminal(self) -> bool:
        """Return True if the state is terminal (no further transitions)."""
        return self in (
            TaskState.COMPLETED,
            TaskState.FAILED,
            TaskState.CANCELED,
            TaskState.REJECTED,
        )

class Role(str, enum.Enum):
    """Sender role for a Message."""

    UNSPECIFIED = "ROLE_UNSPECIFIED"
    USER = "ROLE_USER"
    AGENT = "ROLE_AGENT"

@dataclass
class Part:
    """The smallest unit of content inside a Message or Artifact.

    Exactly one of ``text``, ``raw``, ``url``, or ``data`` should be set,
    corresponding to the ``oneof content`` in the proto definition.

    Attributes:
        text: Plain-text content.
        raw: Base64-encoded binary content.
        url: URL pointing to file content.
        data: Arbitrary structured data (JSON value).
        metadata: Optional key-value metadata.
        filename: Optional filename hint.
        media_type: MIME type of the content.
    """

    text: str | None = None
    raw: str | None = None
    url: str | None = None
    data: Any | None = None
    metadata: dict[str, Any] | None = None
    filename: str | None = None
    media_type: str | None = None

    def to_dict(self) -> dict[str, Any]:
        """Serialize to a camelCase dictionary."""
        return _serialize(self)

    @classmethod
    def from_dict(cls, d: dict[str, Any]) -> "Part":
        """Deserialize from a camelCase dictionary."""
        obj = object.__new__(cls)
        _get = d.get
        obj.text = _get("text")
        obj.raw = _get("raw")
        obj.url = _get("url")
        obj.data = _get("data")
        obj.metadata = _get("metadata")
        obj.filename = _get("filename")
        obj.media_type = _get("mediaType")
        return obj

@dataclass
class Message:
    """A single communication turn between client and agent.

    Attributes:
        message_id: Unique identifier for this message.
        role: The sender role (user or agent).
        parts: Content parts of the message.
        context_id: Optional context grouping identifier.
        task_id: Optional associated task identifier.
        metadata: Optional key-value metadata.
        extensions: Extension URIs active for this message.
        reference_task_ids: Task IDs referenced for additional context.
    """

    message_id: str = ""
    role: Role = Role.USER
    parts: list[Part] = field(default_factory=list)
    context_id: str | None = None
    task_id: str | None = None
    metadata: dict[str, Any] | None = None
    extensions: list[str] | None = None
    reference_task_ids: list[str] | None = None

    def __post_init__(self) -> None:
        if not self.message_id:
            self.message_id = str(uuid.uuid4())

    def to_dict(self) -> dict[str, Any]:
        """Serialize to a camelCase dictionary."""
        return _serialize(self)

    @classmethod
    def from_dict(cls, d: dict[str, Any]) -> "Message":
        """Deserialize from a camelCase dictionary."""
        obj = object.__new__(cls)
        _get = d.get
        msg_id = _get("messageId", "")
        obj.message_id = msg_id if msg_id else str(uuid.uuid4())
        role_val = _get("role", "ROLE_USER")
        obj.role = (
            _ROLE_BY_VALUE.get(role_val) if isinstance(role_val, str) else role_val
        )  # type: ignore[assignment]
        raw_parts = _get("parts")
        obj.parts = [Part.from_dict(p) for p in raw_parts] if raw_parts else []
        obj.context_id = _get("contextId")
        obj.task_id = _get("taskId")
        obj.metadata = _get("metadata")
        obj.extensions = _get("extensions")
        obj.reference_task_ids = _get("referenceTaskIds")
        return obj

@dataclass
class Artifact:
    """An output produced by the agent as a result of task processing.

    Attributes:
        artifact_id: Unique identifier within a task.
        parts: Content parts of the artifact.
        name: Human-readable name.
        description: Human-readable description.
        metadata: Optional key-value metadata.
        extensions: Extension URIs relevant to this artifact.
    """

    artifact_id: str = ""
    parts: list[Part] = field(default_factory=list)
    name: str | None = None
    description: str | None = None
    metadata: dict[str, Any] | None = None
    extensions: list[str] | None = None

    def __post_init__(self) -> None:
        if not self.artifact_id:
            self.artifact_id = str(uuid.uuid4())

    def to_dict(self) -> dict[str, Any]:
        """Serialize to a camelCase dictionary."""
        return _serialize(self)

    @classmethod
    def from_dict(cls, d: dict[str, Any]) -> "Artifact":
        """Deserialize from a camelCase dictionary."""
        obj = object.__new__(cls)
        _get = d.get
        art_id = _get("artifactId", "")
        obj.artifact_id = art_id if art_id else str(uuid.uuid4())
        raw_parts = _get("parts")
        obj.parts = [Part.from_dict(p) for p in raw_parts] if raw_parts else []
        obj.name = _get("name")
        obj.description = _get("description")
        obj.metadata = _get("metadata")
        obj.extensions = _get("extensions")
        return obj

@dataclass
class TaskStatus:
    """Current status of a Task.

    Attributes:
        state: The lifecycle state.
        message: An optional message associated with the status.
        timestamp: ISO 8601 timestamp when the status was recorded.
    """

    state: TaskState = TaskState.SUBMITTED
    message: Message | None = None
    timestamp: str | None = None

    def __post_init__(self) -> None:
        if self.timestamp is None:
            self.timestamp = _now_iso()

    def to_dict(self) -> dict[str, Any]:
        """Serialize to a camelCase dictionary."""
        return _serialize(self)

    @classmethod
    def from_dict(cls, d: dict[str, Any]) -> "TaskStatus":
        """Deserialize from a camelCase dictionary."""
        obj = object.__new__(cls)
        _get = d.get
        state_val = _get("state", "TASK_STATE_UNSPECIFIED")
        obj.state = (
            _TASKSTATE_BY_VALUE.get(state_val)
            if isinstance(state_val, str)
            else state_val
        )  # type: ignore[assignment]
        msg = _get("message")
        obj.message = Message.from_dict(msg) if msg else None
        ts = _get("timestamp")
        obj.timestamp = ts if ts is not None else _now_iso()
        return obj

@dataclass
class Task:
    """The fundamental unit of work managed by A2A.

    Attributes:
        id: Server-generated unique identifier.
        status: Current task status.
        context_id: Optional context grouping identifier.
        artifacts: Output artifacts produced so far.
        history: Message history for the task.
        metadata: Optional key-value metadata.
    """

    id: str = ""
    status: TaskStatus = field(default_factory=TaskStatus)
    context_id: str | None = None
    artifacts: list[Artifact] | None = None
    history: list[Message] | None = None
    metadata: dict[str, Any] | None = None

    def __post_init__(self) -> None:
        if not self.id:
            self.id = str(uuid.uuid4())

    def to_dict(self) -> dict[str, Any]:
        """Serialize to a camelCase dictionary."""
        return _serialize(self)

    @classmethod
    def from_dict(cls, d: dict[str, Any]) -> "Task":
        """Deserialize from a camelCase dictionary."""
        obj = object.__new__(cls)
        _get = d.get
        task_id = _get("id", "")
        obj.id = task_id if task_id else str(uuid.uuid4())
        status_raw = _get("status")
        obj.status = TaskStatus.from_dict(status_raw) if status_raw else TaskStatus()
        obj.context_id = _get("contextId")
        artifacts_raw = _get("artifacts")
        obj.artifacts = (
            [Artifact.from_dict(a) for a in artifacts_raw]
            if artifacts_raw is not None
            else None
        )
        history_raw = _get("history")
        obj.history = (
            [Message.from_dict(m) for m in history_raw]
            if history_raw is not None
            else None
        )
        obj.metadata = _get("metadata")
        return obj

@dataclass
class TaskStatusUpdateEvent:
    """Event indicating a change in task status.

    Attributes:
        task_id: The task that changed.
        context_id: The context the task belongs to.
        status: The new status.
        metadata: Optional metadata.
    """

    task_id: str = ""
    context_id: str = ""
    status: TaskStatus = field(default_factory=TaskStatus)
    metadata: dict[str, Any] | None = None

    def to_dict(self) -> dict[str, Any]:
        """Serialize to a camelCase dictionary."""
        return _serialize(self)

    @classmethod
    def from_dict(cls, d: dict[str, Any]) -> "TaskStatusUpdateEvent":
        """Deserialize from a camelCase dictionary."""
        obj = object.__new__(cls)
        _get = d.get
        obj.task_id = _get("taskId", "")
        obj.context_id = _get("contextId", "")
        status_raw = _get("status")
        obj.status = TaskStatus.from_dict(status_raw) if status_raw else TaskStatus()
        obj.metadata = _get("metadata")
        return obj