Skip to content

refactor: Change Well-Known URI Magic Strings to a Constant defined in the specification #270

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

Closed
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
1 change: 1 addition & 0 deletions .github/actions/spelling/excludes.txt
Original file line number Diff line number Diff line change
Expand Up @@ -90,3 +90,4 @@ noxfile.py
^src/a2a/grpc/
^tests/
.pre-commit-config.yaml
^src/a2a/types.py
5 changes: 3 additions & 2 deletions src/a2a/client/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@
SendStreamingMessageResponse,
SetTaskPushNotificationConfigRequest,
SetTaskPushNotificationConfigResponse,
WellKnownUris,
)
from a2a.utils.telemetry import SpanKind, trace_class

Expand All @@ -40,7 +41,7 @@ def __init__(
self,
httpx_client: httpx.AsyncClient,
base_url: str,
agent_card_path: str = '/.well-known/agent.json',
agent_card_path: str = WellKnownUris.AGENT_CARD_WELL_KNOWN_URI,
):
"""Initializes the A2ACardResolver.

Expand Down Expand Up @@ -184,7 +185,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 = WellKnownUris.AGENT_CARD_WELL_KNOWN_URI,
http_kwargs: dict[str, Any] | None = None,
) -> 'A2AClient':
"""Fetches the public AgentCard and initializes an A2A client.
Expand Down
6 changes: 3 additions & 3 deletions src/a2a/server/apps/jsonrpc/fastapi_app.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@
JSONRPCApplication,
)
from a2a.server.request_handlers.jsonrpc_handler import RequestHandler
from a2a.types import AgentCard
from a2a.types import AgentCard, WellKnownUris


logger = logging.getLogger(__name__)
Expand All @@ -23,36 +23,36 @@
(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 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',
agent_card_url: str = WellKnownUris.AGENT_CARD_WELL_KNOWN_URI,
rpc_url: str = '/',
extended_agent_card_url: str = '/agent/authenticatedExtendedCard',
) -> None:
Expand Down Expand Up @@ -83,7 +83,7 @@

def build(
self,
agent_card_url: str = '/.well-known/agent.json',
agent_card_url: str = WellKnownUris.AGENT_CARD_WELL_KNOWN_URI,
rpc_url: str = '/',
extended_agent_card_url: str = '/agent/authenticatedExtendedCard',
**kwargs: Any,
Expand Down
35 changes: 22 additions & 13 deletions src/a2a/server/apps/jsonrpc/jsonrpc_app.py
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,7 @@
SetTaskPushNotificationConfigRequest,
TaskResubscriptionRequest,
UnsupportedOperationError,
WellKnownUris,
)
from a2a.utils.errors import MethodNotImplementedError

Expand Down Expand Up @@ -299,24 +300,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(
Expand Down Expand Up @@ -424,7 +433,7 @@ async def _handle_get_authenticated_extended_agent_card(
@abstractmethod
def build(
self,
agent_card_url: str = '/.well-known/agent.json',
agent_card_url: str = WellKnownUris.AGENT_CARD_WELL_KNOWN_URI,
rpc_url: str = '/',
**kwargs: Any,
) -> FastAPI | Starlette:
Expand Down
8 changes: 4 additions & 4 deletions src/a2a/server/apps/jsonrpc/starlette_app.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@
JSONRPCApplication,
)
from a2a.server.request_handlers.jsonrpc_handler import RequestHandler
from a2a.types import AgentCard
from a2a.types import AgentCard, WellKnownUris


logger = logging.getLogger(__name__)
Expand All @@ -24,35 +24,35 @@
(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 routes(

Check notice on line 53 in src/a2a/server/apps/jsonrpc/starlette_app.py

View workflow job for this annotation

GitHub Actions / Lint Code Base

Copy/pasted code

see src/a2a/server/apps/jsonrpc/fastapi_app.py (26-52)
self,
agent_card_url: str = '/.well-known/agent.json',
agent_card_url: str = WellKnownUris.AGENT_CARD_WELL_KNOWN_URI,
rpc_url: str = '/',
extended_agent_card_url: str = '/agent/authenticatedExtendedCard',
) -> list[Route]:
Expand Down Expand Up @@ -95,7 +95,7 @@
def add_routes_to_app(
self,
app: Starlette,
agent_card_url: str = '/.well-known/agent.json',
agent_card_url: str = WellKnownUris.AGENT_CARD_WELL_KNOWN_URI,
rpc_url: str = '/',
extended_agent_card_url: str = '/agent/authenticatedExtendedCard',
) -> None:
Expand All @@ -116,7 +116,7 @@

def build(
self,
agent_card_url: str = '/.well-known/agent.json',
agent_card_url: str = WellKnownUris.AGENT_CARD_WELL_KNOWN_URI,
rpc_url: str = '/',
extended_agent_card_url: str = '/agent/authenticatedExtendedCard',
**kwargs: Any,
Expand Down
17 changes: 17 additions & 0 deletions src/a2a/types.py
Original file line number Diff line number Diff line change
Expand Up @@ -921,6 +921,20 @@ class UnsupportedOperationError(BaseModel):
"""


class WellKnownUris(BaseModel):
"""
Well-known URIs used in the A2A protocol.
https://datatracker.ietf.org/doc/html/rfc8615
"""

AGENT_CARD_WELL_KNOWN_URI: Literal['/.well-known/agent.json'] = (
'/.well-known/agent.json'
)
"""
The well-known URI at which an AgentCard should be hosted.
"""


class A2AError(
RootModel[
JSONParseError
Expand Down Expand Up @@ -1568,6 +1582,9 @@ class AgentCard(BaseModel):
- Skills: A set of capabilities the agent can perform
- Default modalities/content types supported by the agent.
- Authentication requirements

The AgentCard SHOULD be hosted at the well-known URI specified by the
`WellKnownUris.AGENT_CARD_WELL_KNOWN_URI` constant ("/.well-known/agent.json").
"""

additionalInterfaces: list[AgentInterface] | None = None
Expand Down
16 changes: 11 additions & 5 deletions tests/client/test_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,7 @@
TaskNotCancelableError,
TaskPushNotificationConfig,
TaskQueryParams,
WellKnownUris,
)


Expand Down Expand Up @@ -127,8 +128,7 @@ async def async_iterable_from_list(

class TestA2ACardResolver:
BASE_URL = 'http://example.com'
AGENT_CARD_PATH = '/.well-known/agent.json'
FULL_AGENT_CARD_URL = f'{BASE_URL}{AGENT_CARD_PATH}'
FULL_AGENT_CARD_URL = f'{BASE_URL}{WellKnownUris.AGENT_CARD_WELL_KNOWN_URI}'
EXTENDED_AGENT_CARD_PATH = (
'/agent/authenticatedExtendedCard' # Default path
)
Expand All @@ -153,20 +153,26 @@ async def test_init_parameters_stored_correctly(
httpx_client=mock_httpx_client,
base_url=base_url,
)
assert resolver_default_path.agent_card_path == '.well-known/agent.json'
assert (
resolver_default_path.agent_card_path
== WellKnownUris.AGENT_CARD_WELL_KNOWN_URI.lstrip('/')
)

@pytest.mark.asyncio
async def test_init_strips_slashes(self, mock_httpx_client: AsyncMock):
resolver = A2ACardResolver(
httpx_client=mock_httpx_client,
base_url='http://example.com/', # With trailing slash
agent_card_path='/.well-known/agent.json/', # With leading/trailing slash
agent_card_path=f'{WellKnownUris.AGENT_CARD_WELL_KNOWN_URI}/', # With leading/trailing slash
)
assert (
resolver.base_url == 'http://example.com'
) # Trailing slash stripped
# constructor lstrips agent_card_path, but keeps trailing if provided
assert resolver.agent_card_path == '.well-known/agent.json/'
assert (
resolver.agent_card_path
== WellKnownUris.AGENT_CARD_WELL_KNOWN_URI.lstrip('/') + '/'
)

@pytest.mark.asyncio
async def test_get_agent_card_success_public_only(
Expand Down
8 changes: 5 additions & 3 deletions tests/server/apps/jsonrpc/test_serialization.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,8 @@
from unittest import mock

import pytest

from pydantic import ValidationError
from starlette.testclient import TestClient

from a2a.server.apps import A2AFastAPIApplication, A2AStarletteApplication
Expand All @@ -10,8 +12,8 @@
AgentCard,
In,
SecurityScheme,
WellKnownUris,
)
from pydantic import ValidationError


@pytest.fixture
Expand Down Expand Up @@ -52,7 +54,7 @@ def test_starlette_agent_card_with_api_key_scheme_alias(
app_instance = A2AStarletteApplication(agent_card_with_api_key, handler)
client = TestClient(app_instance.build())

response = client.get('/.well-known/agent.json')
response = client.get(WellKnownUris.AGENT_CARD_WELL_KNOWN_URI)
assert response.status_code == 200
response_data = response.json()

Expand Down Expand Up @@ -84,7 +86,7 @@ def test_fastapi_agent_card_with_api_key_scheme_alias(
app_instance = A2AFastAPIApplication(agent_card_with_api_key, handler)
client = TestClient(app_instance.build())

response = client.get('/.well-known/agent.json')
response = client.get(WellKnownUris.AGENT_CARD_WELL_KNOWN_URI)
assert response.status_code == 200
response_data = response.json()

Expand Down
7 changes: 4 additions & 3 deletions tests/server/test_integration.py
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,7 @@
TaskStatus,
TextPart,
UnsupportedOperationError,
WellKnownUris,
)
from a2a.utils.errors import MethodNotImplementedError

Expand Down Expand Up @@ -147,7 +148,7 @@ def client(app: A2AStarletteApplication, **kwargs):

def test_agent_card_endpoint(client: TestClient, agent_card: AgentCard):
"""Test the agent card endpoint returns expected data."""
response = client.get('/.well-known/agent.json')
response = client.get(WellKnownUris.AGENT_CARD_WELL_KNOWN_URI)
assert response.status_code == 200
data = response.json()
assert data['name'] == agent_card.name
Expand Down Expand Up @@ -315,7 +316,7 @@ def custom_handler(request):
assert response.json() == {'message': 'Hello'}

# Ensure default routes still work
response = client.get('/.well-known/agent.json')
response = client.get(WellKnownUris.AGENT_CARD_WELL_KNOWN_URI)
assert response.status_code == 200
data = response.json()
assert data['name'] == agent_card.name
Expand All @@ -339,7 +340,7 @@ def custom_handler(request):
assert response.json() == {'message': 'Hello'}

# Ensure default routes still work
response = client.get('/.well-known/agent.json')
response = client.get(WellKnownUris.AGENT_CARD_WELL_KNOWN_URI)
assert response.status_code == 200
data = response.json()
assert data['name'] == agent_card.name
Expand Down
Loading