diff --git a/src/a2a/client/client.py b/src/a2a/client/client.py index e29ef8a7..66a1e49b 100644 --- a/src/a2a/client/client.py +++ b/src/a2a/client/client.py @@ -27,6 +27,9 @@ SetTaskPushNotificationConfigRequest, SetTaskPushNotificationConfigResponse, ) +from a2a.utils.constants import ( + AGENT_CARD_WELL_KNOWN_PATH, +) from a2a.utils.telemetry import SpanKind, trace_class @@ -40,8 +43,8 @@ def __init__( self, httpx_client: httpx.AsyncClient, base_url: str, - agent_card_path: str = '/.well-known/agent.json', - ): + agent_card_path: str = AGENT_CARD_WELL_KNOWN_PATH, + ) -> None: """Initializes the A2ACardResolver. Args: @@ -184,7 +187,7 @@ async def _apply_interceptors( async def get_client_from_agent_card_url( httpx_client: httpx.AsyncClient, base_url: str, - agent_card_path: str = '/.well-known/agent.json', + agent_card_path: str = AGENT_CARD_WELL_KNOWN_PATH, http_kwargs: dict[str, Any] | None = None, ) -> 'A2AClient': """Fetches the public AgentCard and initializes an A2A client. diff --git a/src/a2a/server/apps/jsonrpc/fastapi_app.py b/src/a2a/server/apps/jsonrpc/fastapi_app.py index e961e868..9e1e396d 100644 --- a/src/a2a/server/apps/jsonrpc/fastapi_app.py +++ b/src/a2a/server/apps/jsonrpc/fastapi_app.py @@ -10,6 +10,11 @@ ) 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, + EXTENDED_AGENT_CARD_PATH, +) logger = logging.getLogger(__name__) @@ -29,7 +34,7 @@ def __init__( http_handler: RequestHandler, extended_agent_card: AgentCard | None = None, context_builder: CallContextBuilder | None = None, - ): + ) -> None: """Initializes the A2AStarletteApplication. Args: @@ -52,9 +57,9 @@ def __init__( def add_routes_to_app( self, app: FastAPI, - agent_card_url: str = '/.well-known/agent.json', - rpc_url: str = '/', - extended_agent_card_url: str = '/agent/authenticatedExtendedCard', + agent_card_url: str = AGENT_CARD_WELL_KNOWN_PATH, + rpc_url: str = DEFAULT_RPC_URL, + extended_agent_card_url: str = EXTENDED_AGENT_CARD_PATH, ) -> None: """Adds the routes to the FastAPI application. @@ -83,9 +88,9 @@ async def get_extended_agent_card(request: Request) -> Response: def build( self, - agent_card_url: str = '/.well-known/agent.json', - rpc_url: str = '/', - extended_agent_card_url: str = '/agent/authenticatedExtendedCard', + 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: """Builds and returns the FastAPI application instance. diff --git a/src/a2a/server/apps/jsonrpc/jsonrpc_app.py b/src/a2a/server/apps/jsonrpc/jsonrpc_app.py index 37d28d53..5fa1812e 100644 --- a/src/a2a/server/apps/jsonrpc/jsonrpc_app.py +++ b/src/a2a/server/apps/jsonrpc/jsonrpc_app.py @@ -42,6 +42,10 @@ TaskResubscriptionRequest, UnsupportedOperationError, ) +from a2a.utils.constants import ( + AGENT_CARD_WELL_KNOWN_PATH, + DEFAULT_RPC_URL, +) from a2a.utils.errors import MethodNotImplementedError @@ -109,7 +113,7 @@ def __init__( http_handler: RequestHandler, extended_agent_card: AgentCard | None = None, context_builder: CallContextBuilder | None = None, - ): + ) -> None: """Initializes the A2AStarletteApplication. Args: @@ -299,24 +303,32 @@ async def _process_non_streaming_request( request_obj, context ) case SetTaskPushNotificationConfigRequest(): - handler_result = await self.handler.set_push_notification_config( - request_obj, - context, + handler_result = ( + await self.handler.set_push_notification_config( + request_obj, + context, + ) ) case GetTaskPushNotificationConfigRequest(): - handler_result = await self.handler.get_push_notification_config( - request_obj, - context, + handler_result = ( + await self.handler.get_push_notification_config( + request_obj, + context, + ) ) case ListTaskPushNotificationConfigRequest(): - handler_result = await self.handler.list_push_notification_config( - request_obj, - context, + handler_result = ( + await self.handler.list_push_notification_config( + request_obj, + context, + ) ) case DeleteTaskPushNotificationConfigRequest(): - handler_result = await self.handler.delete_push_notification_config( - request_obj, - context, + handler_result = ( + await self.handler.delete_push_notification_config( + request_obj, + context, + ) ) case _: logger.error( @@ -424,8 +436,8 @@ async def _handle_get_authenticated_extended_agent_card( @abstractmethod def build( self, - agent_card_url: str = '/.well-known/agent.json', - rpc_url: str = '/', + agent_card_url: str = AGENT_CARD_WELL_KNOWN_PATH, + rpc_url: str = DEFAULT_RPC_URL, **kwargs: Any, ) -> FastAPI | Starlette: """Builds and returns the JSONRPC application instance. diff --git a/src/a2a/server/apps/jsonrpc/starlette_app.py b/src/a2a/server/apps/jsonrpc/starlette_app.py index 1826841b..70758739 100644 --- a/src/a2a/server/apps/jsonrpc/starlette_app.py +++ b/src/a2a/server/apps/jsonrpc/starlette_app.py @@ -11,6 +11,11 @@ ) 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, + EXTENDED_AGENT_CARD_PATH, +) logger = logging.getLogger(__name__) @@ -30,7 +35,7 @@ def __init__( http_handler: RequestHandler, extended_agent_card: AgentCard | None = None, context_builder: CallContextBuilder | None = None, - ): + ) -> None: """Initializes the A2AStarletteApplication. Args: @@ -52,9 +57,9 @@ def __init__( def routes( self, - agent_card_url: str = '/.well-known/agent.json', - rpc_url: str = '/', - extended_agent_card_url: str = '/agent/authenticatedExtendedCard', + agent_card_url: str = AGENT_CARD_WELL_KNOWN_PATH, + rpc_url: str = DEFAULT_RPC_URL, + extended_agent_card_url: str = EXTENDED_AGENT_CARD_PATH, ) -> list[Route]: """Returns the Starlette Routes for handling A2A requests. @@ -95,9 +100,9 @@ def 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', + agent_card_url: str = AGENT_CARD_WELL_KNOWN_PATH, + rpc_url: str = DEFAULT_RPC_URL, + extended_agent_card_url: str = EXTENDED_AGENT_CARD_PATH, ) -> None: """Adds the routes to the Starlette application. @@ -116,9 +121,9 @@ def add_routes_to_app( def build( self, - agent_card_url: str = '/.well-known/agent.json', - rpc_url: str = '/', - extended_agent_card_url: str = '/agent/authenticatedExtendedCard', + 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, ) -> Starlette: """Builds and returns the Starlette application instance. diff --git a/src/a2a/utils/__init__.py b/src/a2a/utils/__init__.py index eac4ee17..c2cc97e0 100644 --- a/src/a2a/utils/__init__.py +++ b/src/a2a/utils/__init__.py @@ -5,6 +5,11 @@ new_data_artifact, new_text_artifact, ) +from a2a.utils.constants import ( + AGENT_CARD_WELL_KNOWN_PATH, + DEFAULT_RPC_URL, + EXTENDED_AGENT_CARD_PATH, +) from a2a.utils.helpers import ( append_artifact_to_task, are_modalities_compatible, @@ -24,6 +29,9 @@ __all__ = [ + 'AGENT_CARD_WELL_KNOWN_PATH', + 'DEFAULT_RPC_URL', + 'EXTENDED_AGENT_CARD_PATH', 'append_artifact_to_task', 'are_modalities_compatible', 'build_text_artifact', diff --git a/src/a2a/utils/constants.py b/src/a2a/utils/constants.py new file mode 100644 index 00000000..4bb6c050 --- /dev/null +++ b/src/a2a/utils/constants.py @@ -0,0 +1,5 @@ +"""Constants for well-known URIs used throughout the A2A Python SDK.""" + +AGENT_CARD_WELL_KNOWN_PATH = '/.well-known/agent.json' +EXTENDED_AGENT_CARD_PATH = '/agent/authenticatedExtendedCard' +DEFAULT_RPC_URL = '/'