feat(server): restore FastAPI /docs visibility for A2A routes

Author: martimfasantos

docs/migrations/v1_0/README.md

@@ -465,6 +465,30 @@ app = FastAPI(routes=routes)
 uvicorn.run(app, host=host, port=port)
 ```
 
+`FastAPI(routes=routes)` mounts the A2A endpoints correctly, but FastAPI's OpenAPI generator only enumerates routes that are `fastapi.routing.APIRoute` instances, so the A2A endpoints will not appear in `/docs` or `/openapi.json`. To make them visible in the auto-generated OpenAPI schema — grouped into Agent Card, JSON-RPC, and REST sections — use the `add_a2a_routes_to_fastapi` helper:
+
+```python
+from fastapi import FastAPI
+import uvicorn
+
+from a2a.server.routes import (
+    add_a2a_routes_to_fastapi,
+    create_agent_card_routes,
+    create_jsonrpc_routes,
+    create_rest_routes,
+)
+
+app = FastAPI()
+add_a2a_routes_to_fastapi(
+    app,
+    agent_card_routes=create_agent_card_routes(agent_card),
+    jsonrpc_routes=create_jsonrpc_routes(request_handler, rpc_url='/'),
+    rest_routes=create_rest_routes(request_handler),
+)
+
+uvicorn.run(app, host=host, port=port)
+```
+
 > **Example**: [`a2a-mcp-without-framework/server/__main__.py` in PR #509](https://github.com/a2aproject/a2a-samples/pull/509/files#diff-d15d39ae64c3d4e3a36cc6fb442302caf4e32a6dbd858792e7a4bed180a625ac)
 
 ---

src/a2a/server/routes/__init__.py

@@ -5,13 +5,15 @@
     DefaultServerCallContextBuilder,
     ServerCallContextBuilder,
 )
+from a2a.server.routes.helpers import add_a2a_routes_to_fastapi
 from a2a.server.routes.jsonrpc_routes import create_jsonrpc_routes
 from a2a.server.routes.rest_routes import create_rest_routes
 
 
 __all__ = [
     'DefaultServerCallContextBuilder',
     'ServerCallContextBuilder',
+    'add_a2a_routes_to_fastapi',
     'create_agent_card_routes',
     'create_jsonrpc_routes',
     'create_rest_routes',

src/a2a/server/routes/agent_card_routes.py

@@ -41,6 +41,7 @@ def create_agent_card_routes(
         )
 
     async def _get_agent_card(request: Request) -> Response:
+        """Returns the public AgentCard describing this agent's capabilities, supported transports, and skills."""
         card_to_serve = agent_card
         if card_modifier:
             card_to_serve = await card_modifier(card_to_serve)

src/a2a/server/routes/helpers/__init__.py

@@ -0,0 +1,5 @@
+from a2a.server.routes.helpers.fastapi import add_a2a_routes_to_fastapi
+
+__all__ = [
+    'add_a2a_routes_to_fastapi',
+]

src/a2a/server/routes/helpers/_proto_schema.py

@@ -0,0 +1,119 @@
+"""Proto → JSON Schema helpers shared across transport helpers."""
+
+from typing import Any
+
+from google.protobuf.descriptor import Descriptor, FieldDescriptor
+
+from a2a.types.a2a_pb2 import (
+    SendMessageRequest,
+    TaskPushNotificationConfig,
+)
+
+
+REST_BODY_TYPES: dict[tuple[str, str], type] = {
+    ('/message:send', 'POST'): SendMessageRequest,
+    ('/message:stream', 'POST'): SendMessageRequest,
+    ('/tasks/{id}/pushNotificationConfigs', 'POST'): TaskPushNotificationConfig,
+}
+
+# 64-bit integer types serialize as strings in protojson.
+_PROTO_SCALAR_SCHEMAS: dict[int, dict[str, Any]] = {
+    FieldDescriptor.TYPE_DOUBLE: {'type': 'number'},
+    FieldDescriptor.TYPE_FLOAT: {'type': 'number'},
+    FieldDescriptor.TYPE_INT64: {'type': 'string', 'format': 'int64'},
+    FieldDescriptor.TYPE_UINT64: {'type': 'string', 'format': 'uint64'},
+    FieldDescriptor.TYPE_INT32: {'type': 'integer', 'format': 'int32'},
+    FieldDescriptor.TYPE_FIXED64: {'type': 'string', 'format': 'fixed64'},
+    FieldDescriptor.TYPE_FIXED32: {'type': 'integer', 'format': 'fixed32'},
+    FieldDescriptor.TYPE_BOOL: {'type': 'boolean'},
+    FieldDescriptor.TYPE_STRING: {'type': 'string'},
+    FieldDescriptor.TYPE_BYTES: {'type': 'string', 'format': 'byte'},
+    FieldDescriptor.TYPE_UINT32: {'type': 'integer', 'format': 'uint32'},
+    FieldDescriptor.TYPE_SFIXED32: {'type': 'integer'},
+    FieldDescriptor.TYPE_SFIXED64: {'type': 'string'},
+    FieldDescriptor.TYPE_SINT32: {'type': 'integer'},
+    FieldDescriptor.TYPE_SINT64: {'type': 'string'},
+}
+
+_WELL_KNOWN_SCHEMAS: dict[str, dict[str, Any]] = {
+    'google.protobuf.Timestamp': {'type': 'string', 'format': 'date-time'},
+    'google.protobuf.Duration': {'type': 'string'},
+    'google.protobuf.Struct': {'type': 'object'},
+    'google.protobuf.Value': {},
+    'google.protobuf.ListValue': {'type': 'array', 'items': {}},
+    'google.protobuf.Empty': {'type': 'object'},
+    'google.protobuf.Any': {'type': 'object'},
+    'google.protobuf.FieldMask': {'type': 'string'},
+}
+
+
+def field_schema(
+    field: FieldDescriptor, components: dict[str, Any]
+) -> dict[str, Any]:
+    if field.message_type and field.message_type.GetOptions().map_entry:
+        value_field = field.message_type.fields_by_name['value']
+        return {
+            'type': 'object',
+            'additionalProperties': field_schema(value_field, components),
+        }
+
+    if field.type == FieldDescriptor.TYPE_MESSAGE:
+        item = message_schema(field.message_type, components)
+    elif field.type == FieldDescriptor.TYPE_ENUM:
+        item = {
+            'type': 'string',
+            'enum': [v.name for v in field.enum_type.values],
+        }
+    else:
+        item = dict(_PROTO_SCALAR_SCHEMAS.get(field.type, {'type': 'string'}))
+
+    if field.is_repeated:
+        return {'type': 'array', 'items': item}
+    return item
+
+
+def message_schema(
+    descriptor: Descriptor, components: dict[str, Any]
+) -> dict[str, Any]:
+    """Returns a $ref to descriptor's schema, registering it in components if needed."""
+    if descriptor.full_name in _WELL_KNOWN_SCHEMAS:
+        return dict(_WELL_KNOWN_SCHEMAS[descriptor.full_name])
+
+    name = descriptor.name
+    ref = {'$ref': f'#/components/schemas/{name}'}
+    if name in components:
+        return ref
+
+    # Reserve the slot before recursing so cyclic types terminate.
+    components[name] = {}
+
+    real_oneofs = [o for o in descriptor.oneofs if len(o.fields) > 1]
+    oneof_field_names = {f.name for o in real_oneofs for f in o.fields}
+    base_properties = {
+        f.name: field_schema(f, components)
+        for f in descriptor.fields
+        if f.name not in oneof_field_names
+    }
+
+    if not real_oneofs:
+        components[name] = {'type': 'object', 'properties': base_properties}
+        return ref
+
+    variants: list[dict[str, Any]] = [{}]
+    for oneof in real_oneofs:
+        variants = [
+            {**variant, f.name: field_schema(f, components)}
+            for variant in variants
+            for f in oneof.fields
+        ]
+    components[name] = {
+        'oneOf': [
+            {
+                'type': 'object',
+                'properties': {**base_properties, **variant_props},
+                'required': sorted(set(variant_props) & oneof_field_names),
+            }
+            for variant_props in variants
+        ],
+    }
+    return ref

src/a2a/server/routes/helpers/fastapi.py

@@ -0,0 +1,182 @@
+from typing import TYPE_CHECKING, Any
+
+from a2a.server.routes.helpers._proto_schema import (REST_BODY_TYPES,
+                                                     message_schema)
+from a2a.server.routes.helpers.jsonrpc import \
+    DESCRIPTION as _JSONRPC_DESCRIPTION
+from a2a.server.routes.helpers.jsonrpc import \
+    envelope_schema as _jsonrpc_envelope_schema
+from a2a.utils.constants import PROTOCOL_VERSION_1_0, VERSION_HEADER
+
+if TYPE_CHECKING:
+    from fastapi import FastAPI
+    from starlette.routing import BaseRoute, Route
+
+    _package_fastapi_installed = True
+else:
+    try:
+        from fastapi.routing import APIRoute
+        from starlette.routing import Route, request_response
+
+        class _A2ARoute(APIRoute):
+            """APIRoute that uses Starlette's request_response to bypass FastAPI middleware scope requirements."""
+
+            def __init__(self, *args: Any, **kwargs: Any) -> None:
+                super().__init__(*args, **kwargs)
+                self.app = request_response(self.endpoint)
+
+        _package_fastapi_installed = True
+    except ImportError:
+        Route = Any
+        _A2ARoute = Any
+
+        _package_fastapi_installed = False
+
+
+_AGENT_CARD_TAG = 'A2A: Agent Card'
+_JSONRPC_TAG = 'A2A: JSON-RPC'
+_REST_TAG = 'A2A: REST'
+
+_A2A_VERSION_HEADER = {
+    'in': 'header',
+    'name': VERSION_HEADER,
+    'required': True,
+    'schema': {'type': 'string', 'enum': [PROTOCOL_VERSION_1_0]},
+    'example': PROTOCOL_VERSION_1_0,
+}
+
+
+def _request_body_extra(
+    ref: dict[str, Any], description: str
+) -> dict[str, Any]:
+    return {
+        'requestBody': {
+            'description': description,
+            'required': True,
+            'content': {'application/json': {'schema': ref}},
+        },
+    }
+
+
+def _rest_body_extra(
+    route: 'Route', rest_bodies: dict[tuple[str, str], dict[str, Any]]
+) -> dict[str, Any] | None:
+    methods = route.methods or set()
+    for (suffix, method), extra in rest_bodies.items():
+        if method in methods and route.path.endswith(suffix):
+            return extra
+    return None
+
+
+def _attach_route(
+    app: 'FastAPI',
+    route: 'BaseRoute',
+    tag: str,
+    openapi_extra: dict[str, Any] | None,
+    require_version_header: bool = False,
+) -> None:
+    if not (isinstance(route, Route) and route.methods):
+        app.routes.append(route)
+        return
+    # Drop HEAD: Starlette adds it alongside GET, but FastAPI registers duplicate operation IDs.
+    methods = sorted(m for m in route.methods if m != 'HEAD')
+    if require_version_header:
+        extra = dict(openapi_extra or {})
+        extra.setdefault('parameters', [_A2A_VERSION_HEADER])
+        openapi_extra = extra
+    app.routes.append(
+        _A2ARoute(
+            path=route.path,
+            endpoint=route.endpoint,
+            methods=methods,
+            tags=[tag],
+            openapi_extra=openapi_extra,
+        )
+    )
+
+
+def _install_components(app: 'FastAPI', schemas: dict[str, Any]) -> None:
+    original_openapi = app.openapi
+
+    def _openapi() -> dict[str, Any]:
+        if app.openapi_schema:
+            return app.openapi_schema
+        schema = original_openapi()
+        component_schemas = schema.setdefault('components', {}).setdefault(
+            'schemas', {}
+        )
+        for name, sub_schema in schemas.items():
+            component_schemas.setdefault(name, sub_schema)
+        return schema
+
+    app.openapi = _openapi  # type: ignore[method-assign]
+
+
+def add_a2a_routes_to_fastapi(
+    app: 'FastAPI',
+    *,
+    agent_card_routes: 'list[BaseRoute] | None' = None,
+    jsonrpc_routes: 'list[BaseRoute] | None' = None,
+    rest_routes: 'list[BaseRoute] | None' = None,
+) -> None:
+    """Mounts A2A routes on a FastAPI app and enriches them for ``/docs``.
+
+    Re-registers Starlette routes as ``APIRoute`` instances so they appear in
+    the auto-generated OpenAPI schema, tagged and annotated with proto-derived
+    request-body schemas.
+
+    Usage::
+
+        app = FastAPI()
+        add_a2a_routes_to_fastapi(
+            app,
+            agent_card_routes=create_agent_card_routes(agent_card),
+            jsonrpc_routes=create_jsonrpc_routes(request_handler, rpc_url='/'),
+            rest_routes=create_rest_routes(request_handler),
+        )
+
+    Args:
+        app: The FastAPI application to mount the routes on.
+        agent_card_routes: Routes returned by ``create_agent_card_routes``.
+        jsonrpc_routes: Routes returned by ``create_jsonrpc_routes``.
+        rest_routes: Routes returned by ``create_rest_routes``.
+    """
+    if not _package_fastapi_installed:
+        raise ImportError(
+            'The `fastapi` package is required to use '
+            '`add_a2a_routes_to_fastapi`. Install it alongside '
+            '`a2a-sdk[http-server]`.'
+        )
+
+    components: dict[str, Any] = {}
+    jsonrpc_extra = {
+        'summary': 'A2A JSON-RPC endpoint',
+        'description': _JSONRPC_DESCRIPTION,
+        **_request_body_extra(
+            _jsonrpc_envelope_schema(components), 'A2A JSON-RPC 2.0 request'
+        ),
+    }
+    rest_extras = {
+        key: _request_body_extra(
+            message_schema(cls.DESCRIPTOR, components),
+            f'A2A {cls.__name__}',
+        )
+        for key, cls in REST_BODY_TYPES.items()
+    }
+
+    for route in agent_card_routes or ():
+        _attach_route(app, route, _AGENT_CARD_TAG, openapi_extra=None)
+
+    for route in jsonrpc_routes or ():
+        extra = jsonrpc_extra if isinstance(route, Route) else None
+        _attach_route(app, route, _JSONRPC_TAG, openapi_extra=extra, require_version_header=True)
+
+    for route in rest_routes or ():
+        extra = (
+            _rest_body_extra(route, rest_extras)
+            if isinstance(route, Route)
+            else None
+        )
+        _attach_route(app, route, _REST_TAG, openapi_extra=extra, require_version_header=True)
+
+    _install_components(app, components)