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
38 changes: 29 additions & 9 deletions src/a2a/server/apps/jsonrpc/fastapi_app.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,51 +23,47 @@
(SSE).
"""

def __init__(
self,
agent_card: AgentCard,
http_handler: RequestHandler,
extended_agent_card: AgentCard | None = None,
context_builder: CallContextBuilder | 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 build(
def add_routes_to_app(

Check notice on line 52 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 (27-53)
self,
app: FastAPI,
agent_card_url: str = '/.well-known/agent.json',
rpc_url: str = '/',
extended_agent_card_url: str = '/agent/authenticatedExtendedCard',
**kwargs: Any,
) -> FastAPI:
"""Builds and returns the FastAPI application instance.
) -> None:
"""Adds the routes to the FastAPI application.

Args:
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.
**kwargs: Additional keyword arguments to pass to the FastAPI constructor.

Returns:
A configured FastAPI application instance.
"""
app = FastAPI(**kwargs)

@app.post(rpc_url)
async def handle_a2a_request(request: Request) -> Response:
Expand All @@ -85,4 +81,28 @@
request
)

def build(
self,
agent_card_url: str = '/.well-known/agent.json',
rpc_url: str = '/',
extended_agent_card_url: str = '/agent/authenticatedExtendedCard',
**kwargs: Any,
) -> FastAPI:
"""Builds and returns the FastAPI application instance.

Args:
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.
"""
app = FastAPI(**kwargs)

self.add_routes_to_app(
app, agent_card_url, rpc_url, extended_agent_card_url
)

return app
36 changes: 27 additions & 9 deletions src/a2a/server/apps/jsonrpc/starlette_app.py
Original file line number Diff line number Diff line change
Expand Up @@ -92,6 +92,28 @@ def routes(
)
return app_routes

def add_routes_to_app(
self,
app: Starlette,
agent_card_url: str = '/.well-known/agent.json',
rpc_url: str = '/',
extended_agent_card_url: str = '/agent/authenticatedExtendedCard',
) -> None:
"""Adds the routes to the Starlette application.

Args:
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.
"""
routes = self.routes(
agent_card_url=agent_card_url,
rpc_url=rpc_url,
extended_agent_card_url=extended_agent_card_url,
)
app.routes.extend(routes)

def build(
self,
agent_card_url: str = '/.well-known/agent.json',
Expand All @@ -110,14 +132,10 @@ def build(
Returns:
A configured Starlette application instance.
"""
app_routes = self.routes(
agent_card_url=agent_card_url,
rpc_url=rpc_url,
extended_agent_card_url=extended_agent_card_url,
app = Starlette(**kwargs)

self.add_routes_to_app(
app, agent_card_url, rpc_url, extended_agent_card_url
)
if 'routes' in kwargs:
kwargs['routes'].extend(app_routes)
else:
kwargs['routes'] = app_routes

return Starlette(**kwargs)
return app
Loading