a2a.server.apps.jsonrpc package

Submodules

a2a.server.apps.jsonrpc.fastapi_app module

class a2a.server.apps.jsonrpc.fastapi_app.A2AFastAPI(*, debug: ~typing.Annotated[bool, Doc('\n                Boolean indicating if debug tracebacks should be returned on server\n                errors.\n\n                Read more in the\n                [Starlette docs for Applications](https://www.starlette.dev/applications/#instantiating-the-application).\n                ')] = False, routes: ~typing.Annotated[~typing.List[~starlette.routing.BaseRoute] | None, Doc("\n                **Note**: you probably shouldn't use this parameter, it is inherited\n                from Starlette and supported for compatibility.\n\n                ---\n\n                A list of routes to serve incoming HTTP and WebSocket requests.\n                "), <warnings.deprecated object at 0x7f9f2afd0dd0>] = None, title: ~typing.Annotated[str, Doc('\n                The title of the API.\n\n                It will be added to the generated OpenAPI (e.g. visible at `/docs`).\n\n                Read more in the\n                [FastAPI docs for Metadata and Docs URLs](https://fastapi.tiangolo.com/tutorial/metadata/#metadata-for-api).\n\n                **Example**\n\n                ```python\n                from fastapi import FastAPI\n\n                app = FastAPI(title="ChimichangApp")\n                ```\n                ')] = 'FastAPI', summary: ~typing.Annotated[str | None, Doc('\n                A short summary of the API.\n\n                It will be added to the generated OpenAPI (e.g. visible at `/docs`).\n\n                Read more in the\n                [FastAPI docs for Metadata and Docs URLs](https://fastapi.tiangolo.com/tutorial/metadata/#metadata-for-api).\n\n                **Example**\n\n                ```python\n                from fastapi import FastAPI\n\n                app = FastAPI(summary="Deadpond\'s favorite app. Nuff said.")\n                ```\n                ')] = None, description: ~typing.Annotated[str, Doc('\n                A description of the API. Supports Markdown (using\n                [CommonMark syntax](https://commonmark.org/)).\n\n                It will be added to the generated OpenAPI (e.g. visible at `/docs`).\n\n                Read more in the\n                [FastAPI docs for Metadata and Docs URLs](https://fastapi.tiangolo.com/tutorial/metadata/#metadata-for-api).\n\n                **Example**\n\n                ```python\n                from fastapi import FastAPI\n\n                app = FastAPI(\n                    description="""\n                                ChimichangApp API helps you do awesome stuff. 🚀\n\n                                ## Items\n\n                                You can **read items**.\n\n                                ## Users\n\n                                You will be able to:\n\n                                * **Create users** (_not implemented_).\n                                * **Read users** (_not implemented_).\n\n                                """\n                )\n                ```\n                ')] = '', version: ~typing.Annotated[str, Doc('\n                The version of the API.\n\n                **Note** This is the version of your application, not the version of\n                the OpenAPI specification nor the version of FastAPI being used.\n\n                It will be added to the generated OpenAPI (e.g. visible at `/docs`).\n\n                Read more in the\n                [FastAPI docs for Metadata and Docs URLs](https://fastapi.tiangolo.com/tutorial/metadata/#metadata-for-api).\n\n                **Example**\n\n                ```python\n                from fastapi import FastAPI\n\n                app = FastAPI(version="0.0.1")\n                ```\n                ')] = '0.1.0', openapi_url: ~typing.Annotated[str | None, Doc('\n                The URL where the OpenAPI schema will be served from.\n\n                If you set it to `None`, no OpenAPI schema will be served publicly, and\n                the default automatic endpoints `/docs` and `/redoc` will also be\n                disabled.\n\n                Read more in the\n                [FastAPI docs for Metadata and Docs URLs](https://fastapi.tiangolo.com/tutorial/metadata/#openapi-url).\n\n                **Example**\n\n                ```python\n                from fastapi import FastAPI\n\n                app = FastAPI(openapi_url="/api/v1/openapi.json")\n                ```\n                ')] = '/openapi.json', openapi_tags: ~typing.Annotated[~typing.List[~typing.Dict[str, ~typing.Any]] | None, Doc('\n                A list of tags used by OpenAPI, these are the same `tags` you can set\n                in the *path operations*, like:\n\n                * `@app.get("/users/", tags=["users"])`\n                * `@app.get("/items/", tags=["items"])`\n\n                The order of the tags can be used to specify the order shown in\n                tools like Swagger UI, used in the automatic path `/docs`.\n\n                It\'s not required to specify all the tags used.\n\n                The tags that are not declared MAY be organized randomly or based\n                on the tools\' logic. Each tag name in the list MUST be unique.\n\n                The value of each item is a `dict` containing:\n\n                * `name`: The name of the tag.\n                * `description`: A short description of the tag.\n                    [CommonMark syntax](https://commonmark.org/) MAY be used for rich\n                    text representation.\n                * `externalDocs`: Additional external documentation for this tag. If\n                    provided, it would contain a `dict` with:\n                    * `description`: A short description of the target documentation.\n                        [CommonMark syntax](https://commonmark.org/) MAY be used for\n                        rich text representation.\n                    * `url`: The URL for the target documentation. Value MUST be in\n                        the form of a URL.\n\n                Read more in the\n                [FastAPI docs for Metadata and Docs URLs](https://fastapi.tiangolo.com/tutorial/metadata/#metadata-for-tags).\n\n                **Example**\n\n                ```python\n                from fastapi import FastAPI\n\n                tags_metadata = [\n                    {\n                        "name": "users",\n                        "description": "Operations with users. The **login** logic is also here.",\n                    },\n                    {\n                        "name": "items",\n                        "description": "Manage items. So _fancy_ they have their own docs.",\n                        "externalDocs": {\n                            "description": "Items external docs",\n                            "url": "https://fastapi.tiangolo.com/",\n                        },\n                    },\n                ]\n\n                app = FastAPI(openapi_tags=tags_metadata)\n                ```\n                ')] = None, servers: ~typing.Annotated[~typing.List[~typing.Dict[str, ~typing.Any | str]] | None, Doc('\n                A `list` of `dict`s with connectivity information to a target server.\n\n                You would use it, for example, if your application is served from\n                different domains and you want to use the same Swagger UI in the\n                browser to interact with each of them (instead of having multiple\n                browser tabs open). Or if you want to leave fixed the possible URLs.\n\n                If the servers `list` is not provided, or is an empty `list`, the\n                default value would be a `dict` with a `url` value of `/`.\n\n                Each item in the `list` is a `dict` containing:\n\n                * `url`: A URL to the target host. This URL supports Server Variables\n                and MAY be relative, to indicate that the host location is relative\n                to the location where the OpenAPI document is being served. Variable\n                substitutions will be made when a variable is named in `{`brackets`}`.\n                * `description`: An optional string describing the host designated by\n                the URL. [CommonMark syntax](https://commonmark.org/) MAY be used for\n                rich text representation.\n                * `variables`: A `dict` between a variable name and its value. The value\n                    is used for substitution in the server\'s URL template.\n\n                Read more in the\n                [FastAPI docs for Behind a Proxy](https://fastapi.tiangolo.com/advanced/behind-a-proxy/#additional-servers).\n\n                **Example**\n\n                ```python\n                from fastapi import FastAPI\n\n                app = FastAPI(\n                    servers=[\n                        {"url": "https://stag.example.com", "description": "Staging environment"},\n                        {"url": "https://prod.example.com", "description": "Production environment"},\n                    ]\n                )\n                ```\n                ')] = None, dependencies: ~typing.Annotated[~typing.Sequence[~fastapi.params.Depends] | None, Doc('\n                A list of global dependencies, they will be applied to each\n                *path operation*, including in sub-routers.\n\n                Read more about it in the\n                [FastAPI docs for Global Dependencies](https://fastapi.tiangolo.com/tutorial/dependencies/global-dependencies/).\n\n                **Example**\n\n                ```python\n                from fastapi import Depends, FastAPI\n\n                from .dependencies import func_dep_1, func_dep_2\n\n                app = FastAPI(dependencies=[Depends(func_dep_1), Depends(func_dep_2)])\n                ```\n                ')] = None, default_response_class: ~typing.Annotated[~typing.Type[~starlette.responses.Response], Doc('\n                The default response class to be used.\n\n                Read more in the\n                [FastAPI docs for Custom Response - HTML, Stream, File, others](https://fastapi.tiangolo.com/advanced/custom-response/#default-response-class).\n\n                **Example**\n\n                ```python\n                from fastapi import FastAPI\n                from fastapi.responses import ORJSONResponse\n\n                app = FastAPI(default_response_class=ORJSONResponse)\n                ```\n                ')] = <fastapi.datastructures.DefaultPlaceholder object>, redirect_slashes: ~typing.Annotated[bool, Doc('\n                Whether to detect and redirect slashes in URLs when the client doesn\'t\n                use the same format.\n\n                **Example**\n\n                ```python\n                from fastapi import FastAPI\n\n                app = FastAPI(redirect_slashes=True)  # the default\n\n                @app.get("/items/")\n                async def read_items():\n                    return [{"item_id": "Foo"}]\n                ```\n\n                With this app, if a client goes to `/items` (without a trailing slash),\n                they will be automatically redirected with an HTTP status code of 307\n                to `/items/`.\n                ')] = True, docs_url: ~typing.Annotated[str | None, Doc('\n                The path to the automatic interactive API documentation.\n                It is handled in the browser by Swagger UI.\n\n                The default URL is `/docs`. You can disable it by setting it to `None`.\n\n                If `openapi_url` is set to `None`, this will be automatically disabled.\n\n                Read more in the\n                [FastAPI docs for Metadata and Docs URLs](https://fastapi.tiangolo.com/tutorial/metadata/#docs-urls).\n\n                **Example**\n\n                ```python\n                from fastapi import FastAPI\n\n                app = FastAPI(docs_url="/documentation", redoc_url=None)\n                ```\n                ')] = '/docs', redoc_url: ~typing.Annotated[str | None, Doc('\n                The path to the alternative automatic interactive API documentation\n                provided by ReDoc.\n\n                The default URL is `/redoc`. You can disable it by setting it to `None`.\n\n                If `openapi_url` is set to `None`, this will be automatically disabled.\n\n                Read more in the\n                [FastAPI docs for Metadata and Docs URLs](https://fastapi.tiangolo.com/tutorial/metadata/#docs-urls).\n\n                **Example**\n\n                ```python\n                from fastapi import FastAPI\n\n                app = FastAPI(docs_url="/documentation", redoc_url="redocumentation")\n                ```\n                ')] = '/redoc', swagger_ui_oauth2_redirect_url: ~typing.Annotated[str | None, Doc('\n                The OAuth2 redirect endpoint for the Swagger UI.\n\n                By default it is `/docs/oauth2-redirect`.\n\n                This is only used if you use OAuth2 (with the "Authorize" button)\n                with Swagger UI.\n                ')] = '/docs/oauth2-redirect', swagger_ui_init_oauth: ~typing.Annotated[~typing.Dict[str, ~typing.Any] | None, Doc('\n                OAuth2 configuration for the Swagger UI, by default shown at `/docs`.\n\n                Read more about the available configuration options in the\n                [Swagger UI docs](https://swagger.io/docs/open-source-tools/swagger-ui/usage/oauth2/).\n                ')] = None, middleware: ~typing.Annotated[~typing.Sequence[~starlette.middleware.Middleware] | None, Doc('\n                List of middleware to be added when creating the application.\n\n                In FastAPI you would normally do this with `app.add_middleware()`\n                instead.\n\n                Read more in the\n                [FastAPI docs for Middleware](https://fastapi.tiangolo.com/tutorial/middleware/).\n                ')] = None, exception_handlers: ~typing.Annotated[~typing.Dict[int | ~typing.Type[Exception], ~typing.Callable[[~starlette.requests.Request, ~typing.Any], ~typing.Coroutine[~typing.Any, ~typing.Any, ~starlette.responses.Response]]] | None, Doc('\n                A dictionary with handlers for exceptions.\n\n                In FastAPI, you would normally use the decorator\n                `@app.exception_handler()`.\n\n                Read more in the\n                [FastAPI docs for Handling Errors](https://fastapi.tiangolo.com/tutorial/handling-errors/).\n                ')] = None, on_startup: ~typing.Annotated[~typing.Sequence[~typing.Callable[[], ~typing.Any]] | None, Doc('\n                A list of startup event handler functions.\n\n                You should instead use the `lifespan` handlers.\n\n                Read more in the [FastAPI docs for `lifespan`](https://fastapi.tiangolo.com/advanced/events/).\n                ')] = None, on_shutdown: ~typing.Annotated[~typing.Sequence[~typing.Callable[[], ~typing.Any]] | None, Doc('\n                A list of shutdown event handler functions.\n\n                You should instead use the `lifespan` handlers.\n\n                Read more in the\n                [FastAPI docs for `lifespan`](https://fastapi.tiangolo.com/advanced/events/).\n                ')] = None, lifespan: ~typing.Annotated[~typing.Callable[[~fastapi.applications.AppType], ~contextlib.AbstractAsyncContextManager[None]] | ~typing.Callable[[~fastapi.applications.AppType], ~contextlib.AbstractAsyncContextManager[~collections.abc.Mapping[str, ~typing.Any]]] | None, Doc('\n                A `Lifespan` context manager handler. This replaces `startup` and\n                `shutdown` functions with a single context manager.\n\n                Read more in the\n                [FastAPI docs for `lifespan`](https://fastapi.tiangolo.com/advanced/events/).\n                ')] = None, terms_of_service: ~typing.Annotated[str | None, Doc('\n                A URL to the Terms of Service for your API.\n\n                It will be added to the generated OpenAPI (e.g. visible at `/docs`).\n\n                Read more at the\n                [FastAPI docs for Metadata and Docs URLs](https://fastapi.tiangolo.com/tutorial/metadata/#metadata-for-api).\n\n                **Example**\n\n                ```python\n                app = FastAPI(terms_of_service="http://example.com/terms/")\n                ```\n                ')] = None, contact: ~typing.Annotated[~typing.Dict[str, ~typing.Any | str] | None, Doc('\n                A dictionary with the contact information for the exposed API.\n\n                It can contain several fields.\n\n                * `name`: (`str`) The name of the contact person/organization.\n                * `url`: (`str`) A URL pointing to the contact information. MUST be in\n                    the format of a URL.\n                * `email`: (`str`) The email address of the contact person/organization.\n                    MUST be in the format of an email address.\n\n                It will be added to the generated OpenAPI (e.g. visible at `/docs`).\n\n                Read more at the\n                [FastAPI docs for Metadata and Docs URLs](https://fastapi.tiangolo.com/tutorial/metadata/#metadata-for-api).\n\n                **Example**\n\n                ```python\n                app = FastAPI(\n                    contact={\n                        "name": "Deadpoolio the Amazing",\n                        "url": "http://x-force.example.com/contact/",\n                        "email": "dp@x-force.example.com",\n                    }\n                )\n                ```\n                ')] = None, license_info: ~typing.Annotated[~typing.Dict[str, ~typing.Any | str] | None, Doc('\n                A dictionary with the license information for the exposed API.\n\n                It can contain several fields.\n\n                * `name`: (`str`) **REQUIRED** (if a `license_info` is set). The\n                    license name used for the API.\n                * `identifier`: (`str`) An [SPDX](https://spdx.dev/) license expression\n                    for the API. The `identifier` field is mutually exclusive of the `url`\n                    field. Available since OpenAPI 3.1.0, FastAPI 0.99.0.\n                * `url`: (`str`) A URL to the license used for the API. This MUST be\n                    the format of a URL.\n\n                It will be added to the generated OpenAPI (e.g. visible at `/docs`).\n\n                Read more at the\n                [FastAPI docs for Metadata and Docs URLs](https://fastapi.tiangolo.com/tutorial/metadata/#metadata-for-api).\n\n                **Example**\n\n                ```python\n                app = FastAPI(\n                    license_info={\n                        "name": "Apache 2.0",\n                        "url": "https://www.apache.org/licenses/LICENSE-2.0.html",\n                    }\n                )\n                ```\n                ')] = None, openapi_prefix: ~typing.Annotated[str, Doc('\n                A URL prefix for the OpenAPI URL.\n                '), <warnings.deprecated object at 0x7f9f2aa67f50>] = '', root_path: ~typing.Annotated[str, Doc('\n                A path prefix handled by a proxy that is not seen by the application\n                but is seen by external clients, which affects things like Swagger UI.\n\n                Read more about it at the\n                [FastAPI docs for Behind a Proxy](https://fastapi.tiangolo.com/advanced/behind-a-proxy/).\n\n                **Example**\n\n                ```python\n                from fastapi import FastAPI\n\n                app = FastAPI(root_path="/api/v1")\n                ```\n                ')] = '', root_path_in_servers: ~typing.Annotated[bool, Doc('\n                To disable automatically generating the URLs in the `servers` field\n                in the autogenerated OpenAPI using the `root_path`.\n\n                Read more about it in the\n                [FastAPI docs for Behind a Proxy](https://fastapi.tiangolo.com/advanced/behind-a-proxy/#disable-automatic-server-from-root_path).\n\n                **Example**\n\n                ```python\n                from fastapi import FastAPI\n\n                app = FastAPI(root_path_in_servers=False)\n                ```\n                ')] = True, responses: ~typing.Annotated[~typing.Dict[int | str, ~typing.Dict[str, ~typing.Any]] | None, Doc('\n                Additional responses to be shown in OpenAPI.\n\n                It will be added to the generated OpenAPI (e.g. visible at `/docs`).\n\n                Read more about it in the\n                [FastAPI docs for Additional Responses in OpenAPI](https://fastapi.tiangolo.com/advanced/additional-responses/).\n\n                And in the\n                [FastAPI docs for Bigger Applications](https://fastapi.tiangolo.com/tutorial/bigger-applications/#include-an-apirouter-with-a-custom-prefix-tags-responses-and-dependencies).\n                ')] = None, callbacks: ~typing.Annotated[~typing.List[~starlette.routing.BaseRoute] | None, Doc('\n                OpenAPI callbacks that should apply to all *path operations*.\n\n                It will be added to the generated OpenAPI (e.g. visible at `/docs`).\n\n                Read more about it in the\n                [FastAPI docs for OpenAPI Callbacks](https://fastapi.tiangolo.com/advanced/openapi-callbacks/).\n                ')] = None, webhooks: ~typing.Annotated[~fastapi.routing.APIRouter | None, Doc("\n                Add OpenAPI webhooks. This is similar to `callbacks` but it doesn't\n                depend on specific *path operations*.\n\n                It will be added to the generated OpenAPI (e.g. visible at `/docs`).\n\n                **Note**: This is available since OpenAPI 3.1.0, FastAPI 0.99.0.\n\n                Read more about it in the\n                [FastAPI docs for OpenAPI Webhooks](https://fastapi.tiangolo.com/advanced/openapi-webhooks/).\n                ")] = None, deprecated: ~typing.Annotated[bool | None, Doc("\n                Mark all *path operations* as deprecated. You probably don't need it,\n                but it's available.\n\n                It will be added to the generated OpenAPI (e.g. visible at `/docs`).\n\n                Read more about it in the\n                [FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/).\n                ")] = None, include_in_schema: ~typing.Annotated[bool, Doc("\n                To include (or not) all the *path operations* in the generated OpenAPI.\n                You probably don't need it, but it's available.\n\n                This affects the generated OpenAPI (e.g. visible at `/docs`).\n\n                Read more about it in the\n                [FastAPI docs for Query Parameters and String Validations](https://fastapi.tiangolo.com/tutorial/query-params-str-validations/#exclude-parameters-from-openapi).\n                ")] = True, swagger_ui_parameters: ~typing.Annotated[~typing.Dict[str, ~typing.Any] | None, Doc('\n                Parameters to configure Swagger UI, the autogenerated interactive API\n                documentation (by default at `/docs`).\n\n                Read more about it in the\n                [FastAPI docs about how to Configure Swagger UI](https://fastapi.tiangolo.com/how-to/configure-swagger-ui/).\n                ')] = None, generate_unique_id_function: ~typing.Annotated[~typing.Callable[[~fastapi.routing.APIRoute], str], Doc('\n                Customize the function used to generate unique IDs for the *path\n                operations* shown in the generated OpenAPI.\n\n                This is particularly useful when automatically generating clients or\n                SDKs for your API.\n\n                Read more about it in the\n                [FastAPI docs about how to Generate Clients](https://fastapi.tiangolo.com/advanced/generate-clients/#custom-generate-unique-id-function).\n                ')] = <fastapi.datastructures.DefaultPlaceholder object>, separate_input_output_schemas: ~typing.Annotated[bool, Doc("\n                Whether to generate separate OpenAPI schemas for request body and\n                response body when the results would be more precise.\n\n                This is particularly useful when automatically generating clients.\n\n                For example, if you have a model like:\n\n                ```python\n                from pydantic import BaseModel\n\n                class Item(BaseModel):\n                    name: str\n                    tags: list[str] = []\n                ```\n\n                When `Item` is used for input, a request body, `tags` is not required,\n                the client doesn't have to provide it.\n\n                But when using `Item` for output, for a response body, `tags` is always\n                available because it has a default value, even if it's just an empty\n                list. So, the client should be able to always expect it.\n\n                In this case, there would be two different schemas, one for input and\n                another one for output.\n                ")] = True, openapi_external_docs: ~typing.Annotated[~typing.Dict[str, ~typing.Any] | None, Doc('\n                This field allows you to provide additional external documentation links.\n                If provided, it must be a dictionary containing:\n\n                * `description`: A brief description of the external documentation.\n                * `url`: The URL pointing to the external documentation. The value **MUST**\n                be a valid URL format.\n\n                **Example**:\n\n                ```python\n                from fastapi import FastAPI\n\n                external_docs = {\n                    "description": "Detailed API Reference",\n                    "url": "https://example.com/api-docs",\n                }\n\n                app = FastAPI(openapi_external_docs=external_docs)\n                ```\n                ')] = None, **extra: ~typing.Annotated[~typing.Any, Doc('\n                Extra keyword arguments to be stored in the app, not used by FastAPI\n                anywhere.\n                ')])

Bases: FastAPI

A FastAPI application that adds A2A-specific OpenAPI components.

openapi() → dict[str, Any]

Generates the OpenAPI schema for the application.

class a2a.server.apps.jsonrpc.fastapi_app.A2AFastAPIApplication(agent_card: AgentCard, http_handler: RequestHandler, extended_agent_card: AgentCard | None = None, context_builder: CallContextBuilder | None = None, card_modifier: Callable[[AgentCard], AgentCard] | None = None, extended_card_modifier: Callable[[AgentCard, ServerCallContext], AgentCard] | None = None)

Bases: JSONRPCApplication

A FastAPI application implementing the A2A protocol server endpoints.

Handles incoming JSON-RPC requests, routes them to the appropriate handler methods, and manages response generation including Server-Sent Events (SSE).

add_routes_to_app(app: FastAPI, agent_card_url: str = '/.well-known/agent-card.json', rpc_url: str = '/', extended_agent_card_url: str = '/agent/authenticatedExtendedCard') → None

Adds the routes to the FastAPI application.

Parameters:

  • app – The FastAPI application to add the routes to.

  • 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.

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.

a2a.server.apps.jsonrpc.jsonrpc_app module

class a2a.server.apps.jsonrpc.jsonrpc_app.CallContextBuilder

Bases: ABC

A class for building ServerCallContexts using the Starlette Request.

abstractmethod build(request: Request) → ServerCallContext

Builds a ServerCallContext from a Starlette Request.

class a2a.server.apps.jsonrpc.jsonrpc_app.DefaultCallContextBuilder

Bases: CallContextBuilder

A default implementation of CallContextBuilder.

build(request: Request) → ServerCallContext

Builds a ServerCallContext from a Starlette Request.

Parameters:

request – The incoming Starlette Request object.

Returns:

A ServerCallContext instance populated with user and state information from the request.

class a2a.server.apps.jsonrpc.jsonrpc_app.JSONRPCApplication(agent_card: AgentCard, http_handler: RequestHandler, extended_agent_card: AgentCard | None = None, context_builder: CallContextBuilder | None = None, card_modifier: Callable[[AgentCard], AgentCard] | None = None, extended_card_modifier: Callable[[AgentCard, ServerCallContext], AgentCard] | None = None)

Bases: ABC

Base class for A2A JSONRPC applications.

Handles incoming JSON-RPC requests, routes them to the appropriate handler methods, and manages response generation including Server-Sent Events (SSE).

A2ARequestModel = a2a.types.SendMessageRequest | a2a.types.SendStreamingMessageRequest | a2a.types.GetTaskRequest | a2a.types.CancelTaskRequest | a2a.types.SetTaskPushNotificationConfigRequest | a2a.types.GetTaskPushNotificationConfigRequest | a2a.types.ListTaskPushNotificationConfigRequest | a2a.types.DeleteTaskPushNotificationConfigRequest | a2a.types.TaskResubscriptionRequest | a2a.types.GetAuthenticatedExtendedCardRequest
METHOD_TO_MODEL: dict[str, type[SendMessageRequest | SendStreamingMessageRequest | GetTaskRequest | CancelTaskRequest | SetTaskPushNotificationConfigRequest | GetTaskPushNotificationConfigRequest | ListTaskPushNotificationConfigRequest | DeleteTaskPushNotificationConfigRequest | TaskResubscriptionRequest | GetAuthenticatedExtendedCardRequest]] = {'agent/getAuthenticatedExtendedCard': <class 'a2a.types.GetAuthenticatedExtendedCardRequest'>, 'message/send': <class 'a2a.types.SendMessageRequest'>, 'message/stream': <class 'a2a.types.SendStreamingMessageRequest'>, 'tasks/cancel': <class 'a2a.types.CancelTaskRequest'>, 'tasks/get': <class 'a2a.types.GetTaskRequest'>, 'tasks/pushNotificationConfig/delete': <class 'a2a.types.DeleteTaskPushNotificationConfigRequest'>, 'tasks/pushNotificationConfig/get': <class 'a2a.types.GetTaskPushNotificationConfigRequest'>, 'tasks/pushNotificationConfig/list': <class 'a2a.types.ListTaskPushNotificationConfigRequest'>, 'tasks/pushNotificationConfig/set': <class 'a2a.types.SetTaskPushNotificationConfigRequest'>, 'tasks/resubscribe': <class 'a2a.types.TaskResubscriptionRequest'>}
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.

class a2a.server.apps.jsonrpc.jsonrpc_app.StarletteUserProxy(user: BaseUser)

Bases: User

Adapts the Starlette User class to the A2A user representation.

property is_authenticated: bool

Returns whether the current user is authenticated.

property user_name: str

Returns the user name of the current user.

a2a.server.apps.jsonrpc.starlette_app module

class a2a.server.apps.jsonrpc.starlette_app.A2AStarletteApplication(agent_card: AgentCard, http_handler: RequestHandler, extended_agent_card: AgentCard | None = None, context_builder: CallContextBuilder | None = None, card_modifier: Callable[[AgentCard], AgentCard] | None = None, extended_card_modifier: Callable[[AgentCard, ServerCallContext], AgentCard] | None = None)

Bases: JSONRPCApplication

A Starlette application implementing the A2A protocol server endpoints.

Handles incoming JSON-RPC requests, routes them to the appropriate handler methods, and manages response generation including Server-Sent Events (SSE).

add_routes_to_app(app: Starlette, agent_card_url: str = '/.well-known/agent-card.json', rpc_url: str = '/', extended_agent_card_url: str = '/agent/authenticatedExtendedCard') → None

Adds the routes to the Starlette application.

Parameters:

  • app – The Starlette application to add the routes to.

  • agent_card_url – The URL path for the agent card endpoint.

  • rpc_url – The URL path for the A2A JSON-RPC endpoint (POST requests).

  • extended_agent_card_url – The URL for the authenticated extended agent card endpoint.

build(agent_card_url: str = '/.well-known/agent-card.json', rpc_url: str = '/', extended_agent_card_url: str = '/agent/authenticatedExtendedCard', **kwargs: Any) → Starlette

Builds and returns the Starlette application instance.

Parameters:

  • agent_card_url – The URL path for the agent card endpoint.

  • rpc_url – The URL path for the A2A JSON-RPC endpoint (POST requests).

  • extended_agent_card_url – The URL for the authenticated extended agent card endpoint.

  • **kwargs – Additional keyword arguments to pass to the Starlette constructor.

Returns:

A configured Starlette application instance.

routes(agent_card_url: str = '/.well-known/agent-card.json', rpc_url: str = '/', extended_agent_card_url: str = '/agent/authenticatedExtendedCard') → list[Route]

Returns the Starlette Routes for handling A2A requests.

Parameters:

  • agent_card_url – The URL path for the agent card endpoint.

  • rpc_url – The URL path for the A2A JSON-RPC endpoint (POST requests).

  • extended_agent_card_url – The URL for the authenticated extended agent card endpoint.

Returns:

A list of Starlette Route objects.

Module contents

A2A JSON-RPC Applications.

class a2a.server.apps.jsonrpc.A2AFastAPIApplication(agent_card: AgentCard, http_handler: RequestHandler, extended_agent_card: AgentCard | None = None, context_builder: CallContextBuilder | None = None, card_modifier: Callable[[AgentCard], AgentCard] | None = None, extended_card_modifier: Callable[[AgentCard, ServerCallContext], AgentCard] | None = None)

Bases: JSONRPCApplication

A FastAPI application implementing the A2A protocol server endpoints.

Handles incoming JSON-RPC requests, routes them to the appropriate handler methods, and manages response generation including Server-Sent Events (SSE).

add_routes_to_app(app: FastAPI, agent_card_url: str = '/.well-known/agent-card.json', rpc_url: str = '/', extended_agent_card_url: str = '/agent/authenticatedExtendedCard') → None

Adds the routes to the FastAPI application.

Parameters:

  • app – The FastAPI application to add the routes to.

  • 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.

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.

class a2a.server.apps.jsonrpc.A2AStarletteApplication(agent_card: AgentCard, http_handler: RequestHandler, extended_agent_card: AgentCard | None = None, context_builder: CallContextBuilder | None = None, card_modifier: Callable[[AgentCard], AgentCard] | None = None, extended_card_modifier: Callable[[AgentCard, ServerCallContext], AgentCard] | None = None)

Bases: JSONRPCApplication

A Starlette application implementing the A2A protocol server endpoints.

Handles incoming JSON-RPC requests, routes them to the appropriate handler methods, and manages response generation including Server-Sent Events (SSE).

add_routes_to_app(app: Starlette, agent_card_url: str = '/.well-known/agent-card.json', rpc_url: str = '/', extended_agent_card_url: str = '/agent/authenticatedExtendedCard') → None

Adds the routes to the Starlette application.

Parameters:

  • app – The Starlette application to add the routes to.

  • agent_card_url – The URL path for the agent card endpoint.

  • rpc_url – The URL path for the A2A JSON-RPC endpoint (POST requests).

  • extended_agent_card_url – The URL for the authenticated extended agent card endpoint.

build(agent_card_url: str = '/.well-known/agent-card.json', rpc_url: str = '/', extended_agent_card_url: str = '/agent/authenticatedExtendedCard', **kwargs: Any) → Starlette

Builds and returns the Starlette application instance.

Parameters:

  • agent_card_url – The URL path for the agent card endpoint.

  • rpc_url – The URL path for the A2A JSON-RPC endpoint (POST requests).

  • extended_agent_card_url – The URL for the authenticated extended agent card endpoint.

  • **kwargs – Additional keyword arguments to pass to the Starlette constructor.

Returns:

A configured Starlette application instance.

routes(agent_card_url: str = '/.well-known/agent-card.json', rpc_url: str = '/', extended_agent_card_url: str = '/agent/authenticatedExtendedCard') → list[Route]

Returns the Starlette Routes for handling A2A requests.

Parameters:

  • agent_card_url – The URL path for the agent card endpoint.

  • rpc_url – The URL path for the A2A JSON-RPC endpoint (POST requests).

  • extended_agent_card_url – The URL for the authenticated extended agent card endpoint.

Returns:

A list of Starlette Route objects.

class a2a.server.apps.jsonrpc.CallContextBuilder

Bases: ABC

A class for building ServerCallContexts using the Starlette Request.

abstractmethod build(request: Request) → ServerCallContext

Builds a ServerCallContext from a Starlette Request.

class a2a.server.apps.jsonrpc.DefaultCallContextBuilder

Bases: CallContextBuilder

A default implementation of CallContextBuilder.

build(request: Request) → ServerCallContext

Builds a ServerCallContext from a Starlette Request.

Parameters:

request – The incoming Starlette Request object.

Returns:

A ServerCallContext instance populated with user and state information from the request.

class a2a.server.apps.jsonrpc.JSONRPCApplication(agent_card: AgentCard, http_handler: RequestHandler, extended_agent_card: AgentCard | None = None, context_builder: CallContextBuilder | None = None, card_modifier: Callable[[AgentCard], AgentCard] | None = None, extended_card_modifier: Callable[[AgentCard, ServerCallContext], AgentCard] | None = None)

Bases: ABC

Base class for A2A JSONRPC applications.

Handles incoming JSON-RPC requests, routes them to the appropriate handler methods, and manages response generation including Server-Sent Events (SSE).

A2ARequestModel = a2a.types.SendMessageRequest | a2a.types.SendStreamingMessageRequest | a2a.types.GetTaskRequest | a2a.types.CancelTaskRequest | a2a.types.SetTaskPushNotificationConfigRequest | a2a.types.GetTaskPushNotificationConfigRequest | a2a.types.ListTaskPushNotificationConfigRequest | a2a.types.DeleteTaskPushNotificationConfigRequest | a2a.types.TaskResubscriptionRequest | a2a.types.GetAuthenticatedExtendedCardRequest
METHOD_TO_MODEL: dict[str, type[SendMessageRequest | SendStreamingMessageRequest | GetTaskRequest | CancelTaskRequest | SetTaskPushNotificationConfigRequest | GetTaskPushNotificationConfigRequest | ListTaskPushNotificationConfigRequest | DeleteTaskPushNotificationConfigRequest | TaskResubscriptionRequest | GetAuthenticatedExtendedCardRequest]] = {'agent/getAuthenticatedExtendedCard': <class 'a2a.types.GetAuthenticatedExtendedCardRequest'>, 'message/send': <class 'a2a.types.SendMessageRequest'>, 'message/stream': <class 'a2a.types.SendStreamingMessageRequest'>, 'tasks/cancel': <class 'a2a.types.CancelTaskRequest'>, 'tasks/get': <class 'a2a.types.GetTaskRequest'>, 'tasks/pushNotificationConfig/delete': <class 'a2a.types.DeleteTaskPushNotificationConfigRequest'>, 'tasks/pushNotificationConfig/get': <class 'a2a.types.GetTaskPushNotificationConfigRequest'>, 'tasks/pushNotificationConfig/list': <class 'a2a.types.ListTaskPushNotificationConfigRequest'>, 'tasks/pushNotificationConfig/set': <class 'a2a.types.SetTaskPushNotificationConfigRequest'>, 'tasks/resubscribe': <class 'a2a.types.TaskResubscriptionRequest'>}
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.

class a2a.server.apps.jsonrpc.StarletteUserProxy(user: BaseUser)

Bases: User

Adapts the Starlette User class to the A2A user representation.

property is_authenticated: bool

Returns whether the current user is authenticated.

property user_name: str

Returns the user name of the current user.