Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 0 additions & 4 deletions buf.gen.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,3 @@ plugins:
# Generates *_pb2.pyi files.
- remote: buf.build/protocolbuffers/pyi
out: src/a2a/types
# Generates a2a.swagger.json (OpenAPI v2)
- remote: buf.build/grpc-ecosystem/openapiv2
out: src/a2a/types
opt: json_names_for_fields=true
41 changes: 1 addition & 40 deletions src/a2a/server/apps/jsonrpc/fastapi_app.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,3 @@
import importlib.resources
import json
import logging

from collections.abc import Awaitable, Callable
Expand All @@ -18,62 +16,25 @@
except ImportError:
FastAPI = Any

_package_fastapi_installed = False

from a2a.server.apps.jsonrpc.jsonrpc_app import (
CallContextBuilder,
JSONRPCApplication,
)
from a2a.server.context import ServerCallContext
from a2a.server.request_handlers.request_handler import RequestHandler
from a2a.types.a2a_pb2 import AgentCard
from a2a.utils.constants import (
AGENT_CARD_WELL_KNOWN_PATH,
DEFAULT_RPC_URL,
)


logger = logging.getLogger(__name__)


class A2AFastAPI(FastAPI):
"""A FastAPI application that adds A2A-specific OpenAPI components."""

_a2a_components_added: bool = False

def openapi(self) -> dict[str, Any]:
"""Generates the OpenAPI schema for the application."""
if self.openapi_schema:
return self.openapi_schema

# Try to use the a2a.json schema generated from the proto file
# if available, instead of generating one from the python types.
try:
from a2a import types # noqa: PLC0415

schema_file = importlib.resources.files(types).joinpath('a2a.json')
if schema_file.is_file():
self.openapi_schema = json.loads(
schema_file.read_text(encoding='utf-8')
)
if self.openapi_schema:
return self.openapi_schema
except Exception: # noqa: BLE001
logger.warning(
"Could not load 'a2a.json' from 'a2a.types'. Falling back to auto-generation."
)

openapi_schema = super().openapi()
if not self._a2a_components_added:
# A2ARequest is now a Union type of proto messages, so we can't use
# model_json_schema. Instead, we just mark it as added without
# adding the schema since proto types don't have Pydantic schemas.
# The OpenAPI schema will still be functional for the endpoints.
self._a2a_components_added = True
return openapi_schema


class A2AFastAPIApplication(JSONRPCApplication):

Check notice on line 37 in src/a2a/server/apps/jsonrpc/fastapi_app.py

View workflow job for this annotation

GitHub Actions / Lint Code Base

Copy/pasted code

see src/a2a/server/apps/jsonrpc/starlette_app.py (23-41)
"""A FastAPI application implementing the A2A protocol server endpoints.

Handles incoming JSON-RPC requests, routes them to the appropriate
Expand Down Expand Up @@ -180,7 +141,7 @@
Returns:
A configured FastAPI application instance.
"""
app = A2AFastAPI(**kwargs)
app = FastAPI(**kwargs)

self.add_routes_to_app(app, agent_card_url, rpc_url)

Expand Down
Loading
Loading