Skip to content

refactor: Reduce Duplication in server/apps/jsonrpc #209

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 3 commits into
base: main
Choose a base branch
from
Open
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
51 changes: 6 additions & 45 deletions src/a2a/server/apps/jsonrpc/fastapi_app.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,14 +2,11 @@

from typing import Any

from fastapi import FastAPI, Request, Response
from fastapi import FastAPI

from a2a.server.apps.jsonrpc.jsonrpc_app import (
CallContextBuilder,
JSONRPCApplication,
)
from a2a.server.request_handlers.jsonrpc_handler import RequestHandler
from a2a.types import AgentCard
from a2a.utils.constants import (
AGENT_CARD_WELL_KNOWN_PATH,
DEFAULT_RPC_URL,
Expand All @@ -28,32 +25,6 @@ class A2AFastAPIApplication(JSONRPCApplication):
(SSE).
"""

def __init__(
self,
agent_card: AgentCard,
http_handler: RequestHandler,
extended_agent_card: AgentCard | None = None,
context_builder: CallContextBuilder | None = None,
) -> None:
"""Initializes the A2AStarletteApplication.

Args:
agent_card: The AgentCard describing the agent's capabilities.
http_handler: The handler instance responsible for processing A2A
requests via http.
extended_agent_card: An optional, distinct AgentCard to be served
at the authenticated extended card endpoint.
context_builder: The CallContextBuilder used to construct the
ServerCallContext passed to the http_handler. If None, no
ServerCallContext is passed.
"""
super().__init__(
agent_card=agent_card,
http_handler=http_handler,
extended_agent_card=extended_agent_card,
context_builder=context_builder,
)

def add_routes_to_app(
self,
app: FastAPI,
Expand All @@ -69,22 +40,12 @@ def add_routes_to_app(
rpc_url: The URL for the A2A JSON-RPC endpoint.
extended_agent_card_url: The URL for the authenticated extended agent card endpoint.
"""
route_definitions = self._get_route_definitions(
agent_card_url, rpc_url, extended_agent_card_url
)

@app.post(rpc_url)
async def handle_a2a_request(request: Request) -> Response:
return await self._handle_requests(request)

@app.get(agent_card_url)
async def get_agent_card(request: Request) -> Response:
return await self._handle_get_agent_card(request)

if self.agent_card.supportsAuthenticatedExtendedCard:

@app.get(extended_agent_card_url)
async def get_extended_agent_card(request: Request) -> Response:
return await self._handle_get_authenticated_extended_agent_card(
request
)
for route_def in route_definitions:
app.add_api_route(**route_def)

def build(
self,
Expand Down
39 changes: 38 additions & 1 deletion src/a2a/server/apps/jsonrpc/jsonrpc_app.py
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,7 @@
from a2a.utils.constants import (
AGENT_CARD_WELL_KNOWN_PATH,
DEFAULT_RPC_URL,
EXTENDED_AGENT_CARD_PATH,
)
from a2a.utils.errors import MethodNotImplementedError

Expand Down Expand Up @@ -433,18 +434,54 @@ async def _handle_get_authenticated_extended_agent_card(
status_code=404,
)

def _get_route_definitions(
self,
agent_card_url: str,
rpc_url: str,
extended_agent_card_url: str,
) -> list[dict[str, Any]]:
"""Generates a list of route definitions for the application."""
routes = [
{
'path': rpc_url,
'endpoint': self._handle_requests,
'methods': ['POST'],
'name': 'a2a_handler',
},
{
'path': agent_card_url,
'endpoint': self._handle_get_agent_card,
'methods': ['GET'],
'name': 'agent_card',
},
]

if self.agent_card.supportsAuthenticatedExtendedCard:
routes.append(
{
'path': extended_agent_card_url,
'endpoint': self._handle_get_authenticated_extended_agent_card,
'methods': ['GET'],
'name': 'authenticated_extended_agent_card',
}
)
return routes

@abstractmethod
def build(
self,
agent_card_url: str = AGENT_CARD_WELL_KNOWN_PATH,
rpc_url: str = DEFAULT_RPC_URL,
extended_agent_card_url: str = EXTENDED_AGENT_CARD_PATH,
**kwargs: Any,
) -> FastAPI | Starlette:
"""Builds and returns the JSONRPC application instance.

Args:
agent_card_url: The URL for the agent card endpoint.
rpc_url: The URL for the A2A JSON-RPC 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:
Expand Down
58 changes: 4 additions & 54 deletions src/a2a/server/apps/jsonrpc/starlette_app.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,11 +6,8 @@
from starlette.routing import Route

from a2a.server.apps.jsonrpc.jsonrpc_app import (
CallContextBuilder,
JSONRPCApplication,
)
from a2a.server.request_handlers.jsonrpc_handler import RequestHandler
from a2a.types import AgentCard
from a2a.utils.constants import (
AGENT_CARD_WELL_KNOWN_PATH,
DEFAULT_RPC_URL,
Expand All @@ -29,32 +26,6 @@ class A2AStarletteApplication(JSONRPCApplication):
(SSE).
"""

def __init__(
self,
agent_card: AgentCard,
http_handler: RequestHandler,
extended_agent_card: AgentCard | None = None,
context_builder: CallContextBuilder | None = None,
) -> None:
"""Initializes the A2AStarletteApplication.

Args:
agent_card: The AgentCard describing the agent's capabilities.
http_handler: The handler instance responsible for processing A2A
requests via http.
extended_agent_card: An optional, distinct AgentCard to be served
at the authenticated extended card endpoint.
context_builder: The CallContextBuilder used to construct the
ServerCallContext passed to the http_handler. If None, no
ServerCallContext is passed.
"""
super().__init__(
agent_card=agent_card,
http_handler=http_handler,
extended_agent_card=extended_agent_card,
context_builder=context_builder,
)

def routes(
self,
agent_card_url: str = AGENT_CARD_WELL_KNOWN_PATH,
Expand All @@ -71,31 +42,10 @@ def routes(
Returns:
A list of Starlette Route objects.
"""
app_routes = [
Route(
rpc_url,
self._handle_requests,
methods=['POST'],
name='a2a_handler',
),
Route(
agent_card_url,
self._handle_get_agent_card,
methods=['GET'],
name='agent_card',
),
]

if self.agent_card.supportsAuthenticatedExtendedCard:
app_routes.append(
Route(
extended_agent_card_url,
self._handle_get_authenticated_extended_agent_card,
methods=['GET'],
name='authenticated_extended_agent_card',
)
)
return app_routes
route_definitions = self._get_route_definitions(
agent_card_url, rpc_url, extended_agent_card_url
)
return [Route(**route_def) for route_def in route_definitions]

def add_routes_to_app(
self,
Expand Down
Loading