### FastAPI A2A Application Setup Source: https://a2a-protocol.org/latest/sdk/python/api/a2a.server.apps Implements the A2A protocol server endpoints using FastAPI. It handles incoming JSON-RPC requests, routes them to appropriate handlers, and manages response generation, including Server-Sent Events (SSE). Dependencies include FastAPI and the A2A SDK. ```python from a2a.server.apps.jsonrpc import A2AFastAPIApplication # Assuming agent_card and http_handler are already defined app_instance = A2AFastAPIApplication(agent_card, http_handler) # Add routes to the FastAPI application app_instance.add_routes_to_app(fastapi_app) # Build and return the FastAPI application instance built_app = app_instance.build() ``` -------------------------------- ### Default Request Handler for Push Notification Configuration Source: https://a2a-protocol.org/latest/sdk/python/api/a2a.server Manages default A2A JSON-RPC methods for task push notification configurations, including setting, getting, listing, and deleting. These operations require a PushConfigStore to be configured. ```python async def _on_delete_task_push_notification_config(_params: DeleteTaskPushNotificationConfigParams, _context: ServerCallContext | None = None) -> None: """ Default handler for ‘tasks/pushNotificationConfig/delete’. Requires a PushConfigStore to be configured. """ pass async def _on_get_task_push_notification_config(_params: TaskIdParams | GetTaskPushNotificationConfigParams, _context: ServerCallContext | None = None) -> TaskPushNotificationConfig: """ Default handler for ‘tasks/pushNotificationConfig/get’. Requires a PushConfigStore to be configured. """ pass async def _on_list_task_push_notification_config(_params: ListTaskPushNotificationConfigParams, _context: ServerCallContext | None = None) -> list[TaskPushNotificationConfig]: """ Default handler for ‘tasks/pushNotificationConfig/list’. Requires a PushConfigStore to be configured. """ pass async def _on_set_task_push_notification_config(_params: TaskPushNotificationConfig, _context: ServerCallContext | None = None) -> TaskPushNotificationConfig: """ Default handler for ‘tasks/pushNotificationConfig/set’. Requires a PushNotifier to be configured. """ pass ``` -------------------------------- ### Starlette A2A Application Setup Source: https://a2a-protocol.org/latest/sdk/python/api/a2a.server.apps Implements the A2A protocol server endpoints using Starlette. This class handles JSON-RPC requests, routes them to handlers, and manages response generation, including Server-Sent Events (SSE). It requires Starlette and the A2A SDK. ```python from a2a.server.apps.jsonrpc import A2AStarletteApplication # Assuming agent_card and http_handler are already defined app_instance = A2AStarletteApplication(agent_card, http_handler) # Add routes to the Starlette application app_instance.add_routes_to_app(starlette_app) # Build and return the Starlette application instance built_app = app_instance.build() # Get Starlette Routes for handling A2A requests routes_list = app_instance.routes() ``` -------------------------------- ### REST Get Push Notification Configuration API Source: https://a2a-protocol.org/latest/sdk/python/api/a2a.server Handles the ‘tasks/pushNotificationConfig/get’ REST method. ```APIDOC ## GET /tasks/pushNotificationConfig/get ### Description Handles the ‘tasks/pushNotificationConfig/get’ REST method. ### Method GET ### Endpoint /tasks/pushNotificationConfig/get ### Parameters #### Query Parameters - **request** (Request) - Required - The incoming Request object. - **context** (ServerCallContext) - Required - Context provided by the server. ### Response #### Success Response (200) - **dict[str, Any]** - A dict containing the config. ### Request Example ```http GET /tasks/pushNotificationConfig/get HTTP/1.1 Host: example.com ``` ### Response Example ```json { "config": { ... } } ``` ``` -------------------------------- ### GET /tasks/pushNotificationConfig/get Source: https://a2a-protocol.org/latest/sdk/python/api/a2a.server Handles the ‘tasks/pushNotificationConfig/get’ REST method to retrieve push notification configurations. ```APIDOC ## GET /tasks/pushNotificationConfig/get ### Description Handles the ‘tasks/pushNotificationConfig/get’ REST method to retrieve push notification configurations. ### Method GET ### Endpoint /tasks/pushNotificationConfig/get ### Parameters #### Path Parameters None #### Query Parameters None #### Request Body None ### Request Example ```json { "example": "No request body needed" } ``` ### Response #### Success Response (200) - **dict** - A dict containing the push notification configuration. #### Response Example ```json { "example": "{\"enabled\": true, \"endpoint\": \"http://example.com/push\"}" } ``` ``` -------------------------------- ### FastAPI: OpenAPI Webhooks Source: https://a2a-protocol.org/latest/sdk/python/api/a2a.server.apps Demonstrates how to add OpenAPI webhooks to a FastAPI application. Webhooks are server-initiated asynchronous notifications, independent of specific path operations, and are available starting from OpenAPI 3.1.0 and FastAPI 0.99.0. ```python from typing import Annotated, Any from fastapi import FastAPI, APIRouter, Doc app = FastAPI() webhook_router = APIRouter() app.add_api_route("/webhooks/", webhooks: Annotated[APIRouter | None, Doc( "Add OpenAPI webhooks. This is similar to `callbacks` but it doesn't\n depend on specific *path operations*. " )] = webhook_router) ``` -------------------------------- ### JSONRPCHandler - Get Task Source: https://a2a-protocol.org/latest/sdk/python/api/a2a.server Handles the JSON-RPC method 'tasks/get' to retrieve a task. ```APIDOC ## POST /jsonrpc ### Description Handles the JSON-RPC method 'tasks/get' to retrieve a task. ### Method POST ### Endpoint /jsonrpc ### Parameters #### Request Body - **request** (GetTaskRequest) - Required - The incoming GetTaskRequest object. - **context** (ServerCallContext) - Optional - Context provided by the server. ### Request Example ```json { "method": "tasks/get", "params": { "request": { ... }, "context": { ... } } } ``` ### Response #### Success Response (200) - **response** (GetTaskResponse) - The response object containing the Task or a JSON-RPC error. #### Response Example ```json { "result": { ... }, "id": "some-id" } ``` ``` -------------------------------- ### FastAPI Initialization with Summary - Python Source: https://a2a-protocol.org/latest/sdk/python/api/a2a.server.apps Shows how to initialize a FastAPI application with a short summary. The 'summary' parameter provides a brief description of the API, which appears in the OpenAPI documentation. This is useful for providing a quick overview of the application's purpose. ```python from fastapi import FastAPI app = FastAPI(summary="Deadpond's favorite app. Nuff said.") ``` -------------------------------- ### FastAPI Initialization with Description - Python Source: https://a2a-protocol.org/latest/sdk/python/api/a2a.server.apps Illustrates initializing a FastAPI application with a detailed description that supports Markdown. The 'description' parameter allows for rich text formatting, including headings and lists, enhancing the OpenAPI documentation. This is helpful for providing comprehensive details about the API's functionality. ```python from fastapi import FastAPI app = FastAPI( description=""" ChimichangApp API helps you do awesome stuff. 🚀 ## Items You can **read items**. ## Users You will be able to: * **Create users** (_not implemented_). * **Read users** (_not implemented_). """) ``` -------------------------------- ### GET /message/stream Source: https://a2a-protocol.org/latest/sdk/python/api/a2a.server Handles the ‘message/stream’ REST method to stream messages. ```APIDOC ## GET /message/stream ### Description Handles the ‘message/stream’ REST method to stream messages. Yields response objects as they are produced by the underlying handler’s stream. ### Method GET ### Endpoint /message/stream ### Parameters #### Path Parameters None #### Query Parameters None #### Request Body None ### Request Example ```json { "example": "No request body needed" } ``` ### Response #### Success Response (200) - **AsyncIterator[str]** - Yields JSON serialized objects containing streaming events (Task, Message, TaskStatusUpdateEvent, TaskArtifactUpdateEvent) as JSON. #### Response Example ```json { "example": "{\"type\": \"MessageEvent\", \"content\": \"Received message content\"}" } ``` ``` -------------------------------- ### FastAPI Initialization with OpenAPI Tags - Python Source: https://a2a-protocol.org/latest/sdk/python/api/a2a.server.apps Illustrates how to define OpenAPI tags for organizing API endpoints. The 'openapi_tags' parameter accepts a list of dictionaries, where each dictionary defines a tag's name, description, and optional external documentation URL. This helps in structuring and presenting API documentation effectively. ```python from fastapi import FastAPI tags_metadata = [ { "name": "users", "description": "Operations with users. The **login** logic is also here.", }, { "name": "items", "description": "Manage items. So _fancy_ they have their own docs.", "externalDocs": { "description": "Items external docs", "url": "https://fastapi.tiangolo.com/", }, }, ] app = FastAPI(openapi_tags=tags_metadata) ``` -------------------------------- ### Get Task API Source: https://a2a-protocol.org/latest/sdk/python/api/a2a This API allows you to retrieve details for a specific task. It uses a JSON-RPC request format. ```APIDOC ## POST /tasks/get ### Description Retrieves details for a specific task. ### Method POST ### Endpoint /tasks/get ### Parameters #### Request Body - **id** (str | int) - Required - The identifier for this request. - **jsonrpc** (Literal['2.0']) - Optional, defaults to '2.0' - The version of the JSON-RPC protocol. - **method** (Literal['tasks/get']) - Optional, defaults to 'tasks/get' - The method name. - **params** (TaskQueryParams) - Required - The parameters for querying a task. ### Request Example ```json { "id": "request-2", "method": "tasks/get", "params": { "taskId": "task-uuid-789" } } ``` ### Response #### Success Response (200) - **id** (str | int | None) - The identifier established by the client. - **jsonrpc** (Literal['2.0']) - The version of the JSON-RPC protocol. - **result** (Task) - The result, containing the requested task details. #### Response Example ```json { "id": "request-2", "jsonrpc": "2.0", "result": { "taskId": "task-uuid-789", "status": "completed", "createdAt": "2023-10-27T10:00:00Z" } } ``` #### Error Response (e.g., 400, 500) - Structure will follow JSON-RPC error response format. ``` -------------------------------- ### REST Get Task API Source: https://a2a-protocol.org/latest/sdk/python/api/a2a.server Handles the ‘v1/tasks/{id}’ REST method. ```APIDOC ## GET /v1/tasks/{id} ### Description Handles the ‘v1/tasks/{id}’ REST method. ### Method GET ### Endpoint /v1/tasks/{id} ### Parameters #### Path Parameters - **id** (str) - Required - The ID of the task to retrieve. #### Query Parameters - **request** (Request) - Required - The incoming Request object. - **context** (ServerCallContext) - Required - Context provided by the server. ### Response #### Success Response (200) - **dict[str, Any]** - A Task object containing the Task. ### Request Example ```http GET /v1/tasks/123 HTTP/1.1 Host: example.com ``` ### Response Example ```json { "task": { ... } } ``` ``` -------------------------------- ### FastAPI Initialization with Version - Python Source: https://a2a-protocol.org/latest/sdk/python/api/a2a.server.apps Demonstrates setting the version for a FastAPI application. The 'version' parameter specifies the application's version, which is included in the OpenAPI documentation. This is crucial for version management and API lifecycle tracking. ```python from fastapi import FastAPI app = FastAPI(version="0.0.1") ``` -------------------------------- ### FastAPI: Configure Application Startup/Shutdown Events (Deprecated) Source: https://a2a-protocol.org/latest/sdk/python/api/a2a.server.apps Allows defining sequences of callable functions to be executed during application startup and shutdown. Note that the `lifespan` context manager is the recommended approach for managing these events in newer FastAPI versions. Refer to FastAPI docs for `lifespan`. ```python from typing import Sequence, Callable, Any from fastapi import FastAPI async def startup_event(): pass async def shutdown_event(): pass app = FastAPI( on_startup=[startup_event], on_shutdown=[shutdown_event] ) ``` -------------------------------- ### FastAPI Initialization with OpenAPI URL - Python Source: https://a2a-protocol.org/latest/sdk/python/api/a2a.server.apps Shows how to configure the URL for serving the OpenAPI schema. By setting 'openapi_url', you can customize where the OpenAPI JSON is available. Setting it to None disables the OpenAPI schema endpoint and associated documentation UIs. ```python from fastapi import FastAPI app = FastAPI(openapi_url="/api/v1/openapi.json") ``` -------------------------------- ### A2AFastAPI Initialization Source: https://a2a-protocol.org/latest/sdk/python/api/a2a.server.apps This section outlines the parameters that can be passed during the initialization of the A2AFastAPI class, which inherits from FastAPI and provides additional configuration for JSONRPC services. ```APIDOC ## Class: A2AFastAPI ### Description An extension of FastAPI specifically designed for JSONRPC applications. It allows configuration of various aspects of the API, including debugging, routing, metadata, and OpenAPI specifications. ### Parameters #### `debug` (bool) - Optional - Boolean indicating if debug tracebacks should be returned on server errors. This parameter is inherited from Starlette. #### `routes` (List[BaseRoute] | None) - Optional - A list of routes to serve incoming HTTP and WebSocket requests. This parameter is inherited from Starlette and is generally not recommended for direct use. #### `title` (str) - Optional - Defaults to 'FastAPI'. The title of the API, which appears in the generated OpenAPI documentation (e.g., at `/docs`). #### `summary` (str | None) - Optional - A short summary of the API, displayed in the OpenAPI documentation. #### `description` (str) - Optional - Defaults to ''. A description of the API that supports Markdown syntax. It is displayed in the OpenAPI documentation. #### `version` (str) - Optional - Defaults to '0.1.0'. The version of the application API. This is not the version of OpenAPI or FastAPI. #### `openapi_url` (str | None) - Optional - Defaults to '/openapi.json'. The URL path where the OpenAPI schema will be served. If set to `None`, OpenAPI schema generation and the `/docs` and `/redoc` endpoints are disabled. #### `openapi_tags` (List[Dict[str, Any]] | None) - Optional - A list of tags used for organizing operations in the OpenAPI documentation. Each tag is a dictionary that can include `name`, `description` (supporting Markdown), and `externalDocs`. ### Request Example (Initialization) ```python from a2a.server.apps.jsonrpc.fastapi_app import A2AFastAPI app = A2AFastAPI( title="My A2A JSONRPC API", version="1.0.0", description="This API handles JSONRPC requests for the A2A protocol.", debug=True ) ``` ### Response Example (Not Applicable for Initialization) Initialization of the application class does not produce a direct response in the typical API sense. Configuration is applied internally. ``` -------------------------------- ### Get Task API Source: https://a2a-protocol.org/latest/sdk/python/api/a2a.server Handles the ‘tasks/get’ method to retrieve the state and history of a specific task. ```APIDOC ## GET /tasks/{task_id} ### Description Retrieves the state and history of a specific task. ### Method GET ### Endpoint /tasks/{task_id} ### Parameters #### Path Parameters - **task_id** (string) - Required - The ID of the task to retrieve. #### Query Parameters - **history_length** (integer) - Optional - The number of historical steps to retrieve. #### Request Body None ### Response #### Success Response (200) - **Task** (Task | None) - The Task object if found, otherwise None. #### Response Example ```json { "task_id": "string", "status": "completed", "history": [] } ``` ``` -------------------------------- ### GET /v1/tasks/{id} Source: https://a2a-protocol.org/latest/sdk/python/api/a2a.server Handles the ‘v1/tasks/{id}’ REST method to retrieve a specific task by its ID. ```APIDOC ## GET /v1/tasks/{id} ### Description Handles the ‘v1/tasks/{id}’ REST method to retrieve a specific task by its ID. ### Method GET ### Endpoint /v1/tasks/{id} ### Parameters #### Path Parameters - **id** (string) - Required - The ID of the task to retrieve. #### Query Parameters None #### Request Body None ### Request Example ```json { "example": "No request body needed" } ``` ### Response #### Success Response (200) - **Task** - A Task object containing the Task details. #### Response Example ```json { "example": "{\"taskId\": \"some-task-id\", \"description\": \"Perform a task\"}" } ``` ``` -------------------------------- ### FastAPI App with Global Dependencies Configuration Source: https://a2a-protocol.org/latest/sdk/python/api/a2a.server.apps Sets up a FastAPI application with global dependencies that are applied to every path operation. This simplifies managing common functionalities across all endpoints. Dependencies are provided as a sequence of `Depends` objects. Refer to FastAPI documentation for detailed usage. ```python from fastapi import Depends, FastAPI from .dependencies import func_dep_1, func_dep_2 app = FastAPI(dependencies=[Depends(func_dep_1), Depends(func_dep_2)]) ``` -------------------------------- ### JSONRPCHandler - Get Authenticated Extended Card Source: https://a2a-protocol.org/latest/sdk/python/api/a2a.server Handles the JSON-RPC method 'agent/authenticatedExtendedCard' to retrieve the authenticated extended agent card. ```APIDOC ## POST /jsonrpc ### Description Handles the JSON-RPC method 'agent/authenticatedExtendedCard' to retrieve the authenticated extended agent card. ### Method POST ### Endpoint /jsonrpc ### Parameters #### Request Body - **request** (GetAuthenticatedExtendedCardRequest) - Required - The incoming GetAuthenticatedExtendedCardRequest object. - **context** (ServerCallContext) - Optional - Context provided by the server. ### Request Example ```json { "method": "agent/authenticatedExtendedCard", "params": { "request": { ... }, "context": { ... } } } ``` ### Response #### Success Response (200) - **response** (GetAuthenticatedExtendedCardResponse) - The response object containing the configuration or a JSON-RPC error. #### Response Example ```json { "result": { ... }, "id": "some-id" } ``` ``` -------------------------------- ### JSONRPCHandler - Get Push Notification Configuration Source: https://a2a-protocol.org/latest/sdk/python/api/a2a.server Handles the JSON-RPC method 'tasks/pushNotificationConfig/get' to retrieve a task's push notification configuration. ```APIDOC ## POST /jsonrpc ### Description Handles the JSON-RPC method 'tasks/pushNotificationConfig/get' to retrieve a task's push notification configuration. ### Method POST ### Endpoint /jsonrpc ### Parameters #### Request Body - **request** (GetTaskPushNotificationConfigRequest) - Required - The incoming GetTaskPushNotificationConfigRequest object. - **context** (ServerCallContext) - Optional - Context provided by the server. ### Request Example ```json { "method": "tasks/pushNotificationConfig/get", "params": { "request": { ... }, "context": { ... } } } ``` ### Response #### Success Response (200) - **response** (GetTaskPushNotificationConfigResponse) - The response object containing the configuration or a JSON-RPC error. #### Response Example ```json { "result": { ... }, "id": "some-id" } ``` ``` -------------------------------- ### FastAPI: Configure Application Lifespan Management Source: https://a2a-protocol.org/latest/sdk/python/api/a2a.server.apps Implements application startup and shutdown logic using a single lifespan context manager. This is the modern and recommended approach in FastAPI, replacing separate `on_startup` and `on_shutdown` functions. It provides a cleaner way to manage application lifecycle events. ```python from fastapi import FastAPI from contextlib import asynccontextmanager @asynccontextmanager async def lifespan(app): # Startup logic print("Starting up...") yield # Shutdown logic print("Shutting down...") app = FastAPI(lifespan=lifespan) ``` -------------------------------- ### Get Task Push Notification Configuration API Source: https://a2a-protocol.org/latest/sdk/python/api/a2a This API allows you to retrieve the push notification configuration for a specific task. It uses a JSON-RPC request format. ```APIDOC ## POST /tasks/pushNotificationConfig/get ### Description Retrieves the push notification configuration for a specific task. ### Method POST ### Endpoint /tasks/pushNotificationConfig/get ### Parameters #### Request Body - **id** (str | int) - Required - The identifier for this request. - **jsonrpc** (Literal['2.0']) - Optional, defaults to '2.0' - The version of the JSON-RPC protocol. - **method** (Literal['tasks/pushNotificationConfig/get']) - Optional, defaults to 'tasks/pushNotificationConfig/get' - The method name. - **params** (GetTaskPushNotificationConfigParams | TaskIdParams) - Required - The parameters for getting a push notification configuration. - **id** (str) - Required - The unique identifier (e.g. UUID) of the task. - **metadata** (dict[str, Any] | None) - Optional - Optional metadata associated with the request. - **pushNotificationConfigId** (str | None) - Optional - The ID of the push notification configuration to retrieve. ### Request Example ```json { "id": "request-1", "method": "tasks/pushNotificationConfig/get", "params": { "id": "task-uuid-123", "pushNotificationConfigId": "config-id-456" } } ``` ### Response #### Success Response (200) - **id** (str | int | None) - The identifier established by the client. - **jsonrpc** (Literal['2.0']) - The version of the JSON-RPC protocol. - **result** (TaskPushNotificationConfig) - The result, containing the requested push notification configuration. #### Response Example ```json { "id": "request-1", "jsonrpc": "2.0", "result": { "notificationType": "email", "destination": "user@example.com" } } ``` #### Error Response (e.g., 400, 500) - Structure will follow JSON-RPC error response format. ``` -------------------------------- ### Initialize JSONRPCApplication (Python) Source: https://a2a-protocol.org/latest/sdk/python/api/a2a.server.apps Defines the base class for A2A JSONRPC applications. It handles incoming JSON-RPC requests, routes them, and manages response generation. Requires an agent card, an HTTP handler, and optional context builders or card modifiers. ```python from typing import Callable, Any, Optional from starlette.applications import Starlette from a2a.server.apps.base import ABC, RequestHandler from a2a.types import AgentCard, SendMessageRequest, SendStreamingMessageRequest, GetTaskRequest, CancelTaskRequest, SetTaskPushNotificationConfigRequest, GetTaskPushNotificationConfigRequest, ListTaskPushNotificationConfigRequest, DeleteTaskPushNotificationConfigRequest, TaskResubscriptionRequest, GetAuthenticatedExtendedCardRequest class JSONRPCApplication(ABC): METHOD_TO_MODEL_: dict[str, type[SendMessageRequest | SendStreamingMessageRequest | GetTaskRequest | CancelTaskRequest | SetTaskPushNotificationConfigRequest | GetTaskPushNotificationConfigRequest | ListTaskPushNotificationConfigRequest | DeleteTaskPushNotificationConfigRequest | TaskResubscriptionRequest | GetAuthenticatedExtendedCardRequest]] = { 'agent/getAuthenticatedExtendedCard': GetAuthenticatedExtendedCardRequest, 'message/send': SendMessageRequest, 'message/stream': SendStreamingMessageRequest, 'tasks/cancel': CancelTaskRequest, 'tasks/get': GetTaskRequest, 'tasks/pushNotificationConfig/delete': DeleteTaskPushNotificationConfigRequest, 'tasks/pushNotificationConfig/get': GetTaskPushNotificationConfigRequest, 'tasks/pushNotificationConfig/list': ListTaskPushNotificationConfigRequest, 'tasks/pushNotificationConfig/set': SetTaskPushNotificationConfigRequest, 'tasks/resubscribe': TaskResubscriptionRequest } def __init__(self, _agent_card: AgentCard, _http_handler: RequestHandler, _extended_agent_card: Optional[AgentCard] = None, _context_builder: Optional[CallContextBuilder] = None, _card_modifier: Optional[Callable[[AgentCard], AgentCard]] = None, _extended_card_modifier: Optional[Callable[[AgentCard, ServerCallContext], AgentCard]] = None): # ... initialization logic ... pass @abstractmethod def _build(self, _agent_card_url: str = '/.well-known/agent-card.json', _rpc_url: str = '/', _extended_agent_card_url: str = '/agent/authenticatedExtendedCard', **kwargs: Any) -> Any | Starlette: pass ``` -------------------------------- ### Get Task Push Notification Configuration API Source: https://a2a-protocol.org/latest/sdk/python/api/a2a.server Handles the ‘tasks/pushNotificationConfig/get’ method to retrieve the current push notification configuration for a task. ```APIDOC ## GET /tasks/{task_id}/pushNotificationConfig ### Description Retrieves the current push notification configuration for a task. ### Method GET ### Endpoint /tasks/{task_id}/pushNotificationConfig ### Parameters #### Path Parameters - **task_id** (string) - Required - The ID of the task. #### Query Parameters None #### Request Body None ### Response #### Success Response (200) - **TaskPushNotificationConfig** (TaskPushNotificationConfig) - The TaskPushNotificationConfig for the task. #### Response Example ```json { "task_id": "string", "enabled": true, "webhook_url": "https://example.com/webhook" } ``` ``` -------------------------------- ### JSONRPCApplication: Build JSONRPC Application Instance (Python) Source: https://a2a-protocol.org/latest/sdk/python/api/a2a.server.apps The abstract _build method in JSONRPCApplication is responsible for constructing and returning a configured JSONRPC application instance. It accepts URLs for agent card and RPC endpoints, along with optional arguments for further customization. ```python _abstractmethod _build(_agent_card_url : str = '/.well-known/agent-card.json'_, _rpc_url : str = '/'_, _extended_agent_card_url : str = '/agent/authenticatedExtendedCard'_, _** kwargs: Any_) → Any | Starlette: """Builds and returns the JSONRPC application instance.""" Parameters: * **agent_card_url** – The URL for the agent card endpoint. * **rpc_url** – The URL for the A2A JSON-RPC endpoint. * **extended_agent_card_url** – The URL for the authenticated extended agent card endpoint. * ****kwargs** – Additional keyword arguments to pass to the FastAPI constructor. Returns: A configured JSONRPC application instance. ``` -------------------------------- ### Python: Get task callback configuration Source: https://a2a-protocol.org/latest/sdk/python/api/a2a.client Retrieves the push notification configuration for a specific task. This function is asynchronous and takes parameters for task identification. ```python async def _get_task_callback(self, _request: GetTaskPushNotificationConfigParams, *_, _context: ClientCallContext | None = None) -> TaskPushNotificationConfig: """ Retrieves the push notification configuration for a specific task. """ ``` -------------------------------- ### FastAPI: OpenAPI External Docs Source: https://a2a-protocol.org/latest/sdk/python/api/a2a.server.apps Shows how to provide additional external documentation links in FastAPI's OpenAPI specification using the `openapi_external_docs` parameter. This parameter accepts a dictionary with a description and a URL. ```python from typing import Annotated, Dict, Any from fastapi import FastAPI, Doc external_docs = { "description": "Detailed API Reference", "url": "https://example.com/api-docs", } app = FastAPI(openapi_external_docs: Annotated[Dict[str, Any] | None, Doc( "This field allows you to provide additional external documentation links." )] = external_docs) ``` -------------------------------- ### gRPC Methods Source: https://a2a-protocol.org/latest/sdk/python/api/a2a.server This section details the asynchronous gRPC methods available in the SDK. ```APIDOC ## POST /grpc/SendMessage ### Description Handles the ‘SendMessage’ gRPC method. Processes incoming messages and returns a response. ### Method POST ### Endpoint /grpc/SendMessage ### Parameters #### Request Body - **request** (SendMessageRequest) - Required - The incoming SendMessageRequest object. - **context** (ServicerContext) - Required - Context provided by the server. ### Response #### Success Response (200) - **response** (SendMessageResponse) - The SendMessageResponse object containing the result (Task or Message). #### Error Response - **error** (ServerError) - Throws an error response if a ServerError is raised by the handler. ``` ```APIDOC ## POST /grpc/StreamMessage ### Description Handles the ‘StreamMessage’ gRPC method. Streams response objects as they are produced by the underlying handler’s stream. ### Method POST ### Endpoint /grpc/StreamMessage ### Parameters #### Request Body - **request** (SendMessageRequest) - Required - The incoming SendMessageRequest object. - **context** (ServicerContext) - Required - Context provided by the server. ### Response #### Success Response (200) - **stream** (AsyncIterable[StreamResponse]) - Yields StreamResponse objects containing streaming events (Task, Message, TaskStatusUpdateEvent, TaskArtifactUpdateEvent). #### Error Response - **error** (ServerError) - gRPC error responses if a ServerError is raised. ``` ```APIDOC ## POST /grpc/TaskSubscription ### Description Handles the ‘TaskSubscription’ gRPC method. Streams response objects as they are produced by the underlying handler’s stream. ### Method POST ### Endpoint /grpc/TaskSubscription ### Parameters #### Request Body - **request** (TaskSubscriptionRequest) - Required - The incoming TaskSubscriptionRequest object. - **context** (ServicerContext) - Required - Context provided by the server. ### Response #### Success Response (200) - **stream** (AsyncIterable[StreamResponse]) - Yields StreamResponse objects containing streaming events. #### Error Response - **error** (ServerError) - gRPC error responses if a ServerError is raised. ``` -------------------------------- ### Python: Get task details via REST transport Source: https://a2a-protocol.org/latest/sdk/python/api/a2a.client Asynchronously retrieves the current state and history of a specific task. Requires task query parameters. ```python async def _get_task(self, _request: TaskQueryParams, *_, _context: ClientCallContext | None = None) -> Task: """ Retrieves the current state and history of a specific task. """ ``` -------------------------------- ### Default Request Handler for Tasks Source: https://a2a-protocol.org/latest/sdk/python/api/a2a.server Handles default A2A JSON-RPC methods for task management, such as canceling, getting, and resubscribing to tasks. These methods interact with the AgentExecutor and TaskStore. ```python async def _on_cancel_task(_params: TaskIdParams, _context: ServerCallContext | None = None) -> Task | None: """ Default handler for ‘tasks/cancel’. Attempts to cancel the task managed by the AgentExecutor. """ pass async def _on_get_task(_params: TaskQueryParams, _context: ServerCallContext | None = None) -> Task | None: """ Default handler for ‘tasks/get’. """ pass def on_resubscribe_to_task(_params: TaskIdParams, _context: ServerCallContext | None = None) -> AsyncGenerator[Message | Task | TaskStatusUpdateEvent | TaskArtifactUpdateEvent]: """ Default handler for ‘tasks/resubscribe’. Allows a client to re-attach to a running streaming task’s event stream. Requires the task and its queue to still be active. """ pass ``` -------------------------------- ### POST /a2a/server/request_handlers/prepare_response_object Source: https://a2a-protocol.org/latest/sdk/python/api/a2a.server Helper method to build appropriate JSONRPCResponse object for RPC methods. ```APIDOC ## POST /a2a/server/request_handlers/prepare_response_object ### Description Helper method to build appropriate JSONRPCResponse object for RPC methods. Based on the type of the response object received from the handler, it constructs either a success response wrapped in the appropriate payload type or an error response. ### Method POST ### Endpoint /a2a/server/request_handlers/prepare_response_object ### Parameters #### Request Body - **request_id** (string | integer) - Required - The ID of the request. - **response** (object) - Required - The object received from the request handler. - **success_response_types** (tuple) - Required - A tuple of expected Pydantic model types for a successful result. - **success_payload_type** (type) - Required - The Pydantic model type for the success payload (e.g., SendMessageSuccessResponse). - **response_type** (type) - Required - The Pydantic RootModel type that wraps the final response (e.g., SendMessageResponse). ### Request Example ```json { "request_id": "req456", "response": { "task_id": "task101", "status": "COMPLETED" }, "success_response_types": ["TaskStatusUpdateEvent"], "success_payload_type": "TaskStatusUpdateSuccessPayload", "response_type": "TaskStatusUpdateResponse" } ``` ### Response #### Success Response (200) - **RT** (object) - A Pydantic model representing the final JSON-RPC response (success or error). #### Response Example ```json { "jsonrpc": "2.0", "result": { "task_id": "task101", "status": "COMPLETED" }, "id": "req456" } ``` ``` -------------------------------- ### FastAPI App with Additional Servers Configuration Source: https://a2a-protocol.org/latest/sdk/python/api/a2a.server.apps Configures a FastAPI application with a list of servers, allowing for multiple domain configurations for Swagger UI interaction. Each server entry includes a URL and an optional description. If no servers are provided, a default URL of '/' is used. This is useful for applications served from different domains. ```python from fastapi import FastAPI app = FastAPI( servers=[ {"url": "https://stag.example.com", "description": "Staging environment"}, {"url": "https://prod.example.com", "description": "Production environment"}, ] ) ``` -------------------------------- ### StarletteUserProxy: Get User Name (Python) Source: https://a2a-protocol.org/latest/sdk/python/api/a2a.server.apps The StarletteUserProxy class includes a _user_name property that retrieves the username of the current user. This property is part of the adaptation of the Starlette User class to the A2A user model. ```python _property _user_name _: str_ """Returns the user name of the current user.""" ``` -------------------------------- ### Build FastAPI Application for A2A Protocol Source: https://a2a-protocol.org/latest/sdk/python/api/a2a.server.apps Builds and returns a FastAPI application instance configured for the A2A protocol. Allows customization of agent card, JSON-RPC, and extended agent card URLs. Accepts additional keyword arguments for the FastAPI constructor. ```python def build( _agent_card_url: str = '/.well-known/agent-card.json', _rpc_url: str = '/', _extended_agent_card_url: str = '/agent/authenticatedExtendedCard', **kwargs: Any ) -> FastAPI: """ Builds and returns the FastAPI application instance. Parameters: agent_card_url – The URL for the agent card endpoint. rpc_url – The URL for the A2A JSON-RPC endpoint. extended_agent_card_url – The URL for the authenticated extended agent card endpoint. **kwargs – Additional keyword arguments to pass to the FastAPI constructor. Returns: A configured FastAPI application instance. """ pass ``` -------------------------------- ### FastAPI Initialization with Title - Python Source: https://a2a-protocol.org/latest/sdk/python/api/a2a.server.apps Demonstrates how to initialize a FastAPI application with a custom title. The 'title' parameter is used to set the API's title, which is visible in the generated OpenAPI documentation (e.g., at '/docs'). ```python from fastapi import FastAPI app = FastAPI(title="ChimichangApp") ``` -------------------------------- ### CallContextBuilder Source: https://a2a-protocol.org/latest/sdk/python/api/a2a.server.apps Abstract base class for building ServerCallContexts using the Starlette Request. ```APIDOC ## CallContextBuilder ### Description An abstract base class for building ServerCallContexts using the Starlette Request. ### Abstract Methods #### `_build(_request: Request) -> ServerCallContext` Builds a ServerCallContext from a Starlette Request. ``` -------------------------------- ### FastAPI App with Custom Docs and Disabled ReDoc URLs Source: https://a2a-protocol.org/latest/sdk/python/api/a2a.server.apps Customizes the URL paths for FastAPI's automatic API documentation. The `docs_url` parameter sets the path for Swagger UI, and `redoc_url` can be set to `None` to disable ReDoc. This allows for flexible management of documentation endpoints. Note that if `openapi_url` is `None`, both `docs_url` and `redoc_url` are disabled. ```python from fastapi import FastAPI app = FastAPI(docs_url="/documentation", redoc_url=None) ``` -------------------------------- ### JSON-RPC Methods Source: https://a2a-protocol.org/latest/sdk/python/api/a2a.server This section details the JSON-RPC methods available for interacting with the server, primarily for managing tasks and notifications. ```APIDOC ## POST /jsonrpc ### Description Handles various JSON-RPC methods for task and notification management, agent authentication, and context manipulation. ### Method POST ### Endpoint /jsonrpc ### Parameters #### Request Body - **request** (object) - Required - The JSON-RPC request object, containing method and parameters. - **method** (string) - The JSON-RPC method to call (e.g., 'tasks/pushNotificationConfig/list', 'agent/authenticatedExtendedCard'). - **params** (object) - The parameters for the specified method. - **context** (ServerCallContext | None) - Optional - Context provided by the server. ### Request Example ```json { "method": "tasks/pushNotificationConfig/get", "params": { "request": { "taskId": "task-123" } } } ``` ### Response #### Success Response (200) - **response** (object) - The JSON-RPC response object, containing the result or an error. #### Error Response - **error** (object) - A JSON-RPC error object if the method call fails. ``` ```APIDOC ## POST /jsonrpc/tasks/pushNotificationConfig/delete ### Description Handles the JSON-RPC method for deleting task push notification configurations. ### Method POST ### Endpoint /jsonrpc/tasks/pushNotificationConfig/delete ### Parameters #### Request Body - **request** (DeleteTaskPushNotificationConfigRequest) - Required - The incoming DeleteTaskPushNotificationConfigRequest object. - **context** (ServerCallContext | None) - Optional - Context provided by the server. ### Response #### Success Response (200) - **response** (DeleteTaskPushNotificationConfigResponse) - A DeleteTaskPushNotificationConfigResponse object containing the config. #### Error Response - **error** (object) - A JSON-RPC error object if a ServerError is raised. ``` ```APIDOC ## POST /jsonrpc/agent/authenticatedExtendedCard ### Description Handles the JSON-RPC method for retrieving an authenticated extended agent card. ### Method POST ### Endpoint /jsonrpc/agent/authenticatedExtendedCard ### Parameters #### Request Body - **request** (GetAuthenticatedExtendedCardRequest) - Required - The incoming GetAuthenticatedExtendedCardRequest object. - **context** (ServerCallContext | None) - Optional - Context provided by the server. ### Response #### Success Response (200) - **response** (GetAuthenticatedExtendedCardResponse) - A GetAuthenticatedExtendedCardResponse object containing the config. #### Error Response - **error** (object) - A JSON-RPC error object if a ServerError is raised. ``` ```APIDOC ## POST /jsonrpc/tasks/pushNotificationConfig/get ### Description Handles the JSON-RPC method for retrieving a task push notification configuration. ### Method POST ### Endpoint /jsonrpc/tasks/pushNotificationConfig/get ### Parameters #### Request Body - **request** (GetTaskPushNotificationConfigRequest) - Required - The incoming GetTaskPushNotificationConfigRequest object. - **context** (ServerCallContext | None) - Optional - Context provided by the server. ### Response #### Success Response (200) - **response** (GetTaskPushNotificationConfigResponse) - A GetTaskPushNotificationConfigResponse object containing the config. #### Error Response - **error** (object) - A JSON-RPC error object if a ServerError is raised. ``` ```APIDOC ## POST /jsonrpc/tasks/pushNotificationConfig/list ### Description Handles the JSON-RPC method for listing task push notification configurations. ### Method POST ### Endpoint /jsonrpc/tasks/pushNotificationConfig/list ### Parameters #### Request Body - **request** (ListTaskPushNotificationConfigRequest) - Required - The incoming ListTaskPushNotificationConfigRequest object. - **context** (ServerCallContext | None) - Optional - Context provided by the server. ### Response #### Success Response (200) - **response** (ListTaskPushNotificationConfigResponse) - A ListTaskPushNotificationConfigResponse object containing the config. #### Error Response - **error** (object) - A JSON-RPC error object if a ServerError is raised. ``` ```APIDOC ## POST /jsonrpc/tasks/cancel ### Description Handles the JSON-RPC method for cancelling a task. ### Method POST ### Endpoint /jsonrpc/tasks/cancel ### Parameters #### Request Body - **request** (CancelTaskRequest) - Required - The incoming CancelTaskRequest object. - **context** (ServerCallContext | None) - Optional - Context provided by the server. ### Response #### Success Response (200) - **response** (CancelTaskResponse) - A CancelTaskResponse object containing the updated Task. #### Error Response - **error** (object) - A JSON-RPC error object if a ServerError is raised. ``` ```APIDOC ## POST /jsonrpc/tasks/get ### Description Handles the JSON-RPC method for retrieving a task. ### Method POST ### Endpoint /jsonrpc/tasks/get ### Parameters #### Request Body - **request** (GetTaskRequest) - Required - The incoming GetTaskRequest object. - **context** (ServerCallContext | None) - Optional - Context provided by the server. ### Response #### Success Response (200) - **response** (GetTaskResponse) - A GetTaskResponse object containing the Task. #### Error Response - **error** (object) - A JSON-RPC error object if a ServerError is raised. ``` -------------------------------- ### FastAPI: Swagger UI Parameters Source: https://a2a-protocol.org/latest/sdk/python/api/a2a.server.apps Shows how to configure Swagger UI parameters in FastAPI to customize the interactive API documentation. This allows for adjustments to the appearance and behavior of the Swagger UI interface. ```python from typing import Annotated, Dict, Any from fastapi import FastAPI, Doc app = FastAPI(swagger_ui_parameters: Annotated[Dict[str, Any] | None, Doc( "Parameters to configure Swagger UI, the autogenerated interactive API\n documentation (by default at `/docs`)." )] = None) ``` -------------------------------- ### Default Call Context Builder Implementation Source: https://a2a-protocol.org/latest/sdk/python/api/a2a.server.apps Provides a default implementation for building ServerCallContexts from a Starlette Request. This class inherits from the abstract CallContextBuilder and is used to populate context information from incoming requests. ```python from a2a.server.apps.jsonrpc import DefaultCallContextBuilder # Assuming 'request' is an instance of Starlette's Request context_builder = DefaultCallContextBuilder() server_context = context_builder.build(request) ``` -------------------------------- ### POST /tasks/pushNotificationConfig/list Source: https://a2a-protocol.org/latest/sdk/python/api/a2a.server Handles the ‘tasks/pushNotificationConfig/list’ method. Retrieves the current push notification configurations for a task. ```APIDOC ## POST /tasks/pushNotificationConfig/list ### Description Handles the ‘tasks/pushNotificationConfig/list’ method. Retrieves the current push notification configurations for a task. ### Method POST ### Endpoint /tasks/pushNotificationConfig/list ### Parameters #### Request Body - **params** (ListTaskPushNotificationConfigParams) - Required - Parameters including the task ID. - **context** (ServerCallContext | None) - Optional - Context provided by the server. ### Response #### Success Response (200) - **Returns** - The list[TaskPushNotificationConfig] for the task. ### Response Example ```json [ { "task_id": "task_jkl", "notification_url": "http://example.com/notify1" }, { "task_id": "task_jkl", "notification_url": "http://example.com/notify2" } ] ``` ``` -------------------------------- ### FastAPI: Additional Responses in OpenAPI Source: https://a2a-protocol.org/latest/sdk/python/api/a2a.server.apps Shows how to define and include additional responses in the OpenAPI specification for path operations. This allows for more detailed documentation of possible API responses, visible in tools like Swagger UI. ```python from typing import Annotated, Dict, Any from fastapi import FastAPI, Doc app = FastAPI() @app.get("/items/", response_model=dict, responses: Annotated[Dict[int | str, Dict[str, Any]] | None, Doc( "Additional responses to be shown in OpenAPI." )] = None) ```