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
7 changes: 7 additions & 0 deletions src/a2a/server/apps/jsonrpc/fastapi_app.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
AGENT_CARD_WELL_KNOWN_PATH,
DEFAULT_RPC_URL,
EXTENDED_AGENT_CARD_PATH,
PREV_AGENT_CARD_WELL_KNOWN_PATH,
)


Expand Down Expand Up @@ -89,6 +90,12 @@ def add_routes_to_app(
)(self._handle_requests)
app.get(agent_card_url)(self._handle_get_agent_card)

# add deprecated path only if the agent_card_url uses default well-known path
if agent_card_url == AGENT_CARD_WELL_KNOWN_PATH:
app.get(PREV_AGENT_CARD_WELL_KNOWN_PATH, include_in_schema=False)(
self.handle_deprecated_agent_card_path
)

if self.agent_card.supports_authenticated_extended_card:
app.get(extended_agent_card_url)(
self._handle_get_authenticated_extended_agent_card
Expand Down
10 changes: 10 additions & 0 deletions src/a2a/server/apps/jsonrpc/jsonrpc_app.py
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,7 @@
AGENT_CARD_WELL_KNOWN_PATH,
DEFAULT_RPC_URL,
EXTENDED_AGENT_CARD_PATH,
PREV_AGENT_CARD_WELL_KNOWN_PATH,
)
from a2a.utils.errors import MethodNotImplementedError

Expand Down Expand Up @@ -436,6 +437,15 @@ async def _handle_get_agent_card(self, request: Request) -> JSONResponse:
)
)

async def handle_deprecated_agent_card_path(
self, request: Request
) -> JSONResponse:
"""Handles GET requests for the deprecated agent card endpoint."""
logger.warning(
f"Deprecated agent card endpoint '{PREV_AGENT_CARD_WELL_KNOWN_PATH}' accessed. Please use '{AGENT_CARD_WELL_KNOWN_PATH}' instead. This endpoint will be removed in a future version."
)
return await self._handle_get_agent_card(request)

async def _handle_get_authenticated_extended_agent_card(
self, request: Request
) -> JSONResponse:
Expand Down
12 changes: 12 additions & 0 deletions src/a2a/server/apps/jsonrpc/starlette_app.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@
AGENT_CARD_WELL_KNOWN_PATH,
DEFAULT_RPC_URL,
EXTENDED_AGENT_CARD_PATH,
PREV_AGENT_CARD_WELL_KNOWN_PATH,
)


Expand Down Expand Up @@ -86,6 +87,17 @@ def routes(
),
]

# add deprecated path only if the agent_card_url uses default well-known path
if agent_card_url == AGENT_CARD_WELL_KNOWN_PATH:
app_routes.append(
Route(
PREV_AGENT_CARD_WELL_KNOWN_PATH,
self.handle_deprecated_agent_card_path,
methods=['GET'],
name='agent_card_path_deprecated',
)
)

if self.agent_card.supports_authenticated_extended_card:
app_routes.append(
Route(
Expand Down
2 changes: 2 additions & 0 deletions src/a2a/utils/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@
AGENT_CARD_WELL_KNOWN_PATH,
DEFAULT_RPC_URL,
EXTENDED_AGENT_CARD_PATH,
PREV_AGENT_CARD_WELL_KNOWN_PATH,
)
from a2a.utils.helpers import (
append_artifact_to_task,
Expand All @@ -34,6 +35,7 @@
'AGENT_CARD_WELL_KNOWN_PATH',
'DEFAULT_RPC_URL',
'EXTENDED_AGENT_CARD_PATH',
'PREV_AGENT_CARD_WELL_KNOWN_PATH',
'append_artifact_to_task',
'are_modalities_compatible',
'build_text_artifact',
Expand Down
3 changes: 2 additions & 1 deletion src/a2a/utils/constants.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
"""Constants for well-known URIs used throughout the A2A Python SDK."""

AGENT_CARD_WELL_KNOWN_PATH = '/.well-known/agent.json'
AGENT_CARD_WELL_KNOWN_PATH = '/.well-known/agent-card.json'
PREV_AGENT_CARD_WELL_KNOWN_PATH = '/.well-known/agent.json'
EXTENDED_AGENT_CARD_PATH = '/agent/authenticatedExtendedCard'
DEFAULT_RPC_URL = '/'
8 changes: 6 additions & 2 deletions tests/client/test_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,7 @@
TaskPushNotificationConfig,
TaskQueryParams,
)
from a2a.utils import AGENT_CARD_WELL_KNOWN_PATH


AGENT_CARD = AgentCard(
Expand Down Expand Up @@ -128,7 +129,7 @@ async def async_iterable_from_list(

class TestA2ACardResolver:
BASE_URL = 'http://example.com'
AGENT_CARD_PATH = '/.well-known/agent.json'
AGENT_CARD_PATH = AGENT_CARD_WELL_KNOWN_PATH
FULL_AGENT_CARD_URL = f'{BASE_URL}{AGENT_CARD_PATH}'
EXTENDED_AGENT_CARD_PATH = (
'/agent/authenticatedExtendedCard' # Default path
Expand All @@ -154,7 +155,10 @@ 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
== AGENT_CARD_WELL_KNOWN_PATH
)

@pytest.mark.asyncio
async def test_init_strips_slashes(self, mock_httpx_client: AsyncMock):
Expand Down
85 changes: 70 additions & 15 deletions tests/server/test_integration.py
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,10 @@
TextPart,
UnsupportedOperationError,
)
from a2a.utils import (
AGENT_CARD_WELL_KNOWN_PATH,
PREV_AGENT_CARD_WELL_KNOWN_PATH,
)
from a2a.utils.errors import MethodNotImplementedError


Expand Down Expand Up @@ -147,7 +151,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(AGENT_CARD_WELL_KNOWN_PATH)
assert response.status_code == 200
data = response.json()
assert data['name'] == agent_card.name
Expand All @@ -169,6 +173,36 @@ def test_authenticated_extended_agent_card_endpoint_not_supported(
assert response.status_code == 404 # Starlette's default for no route


def test_agent_card_default_endpoint_has_deprecated_route(
agent_card: AgentCard, handler: mock.AsyncMock
):
"""Test agent card deprecated route is available for default route."""
app_instance = A2AStarletteApplication(agent_card, handler)
client = TestClient(app_instance.build())
response = client.get(AGENT_CARD_WELL_KNOWN_PATH)
assert response.status_code == 200
data = response.json()
assert data['name'] == agent_card.name
response = client.get(PREV_AGENT_CARD_WELL_KNOWN_PATH)
assert response.status_code == 200
data = response.json()
assert data['name'] == agent_card.name


def test_agent_card_custom_endpoint_has_no_deprecated_route(
agent_card: AgentCard, handler: mock.AsyncMock
):
"""Test agent card deprecated route is not available for custom route."""
app_instance = A2AStarletteApplication(agent_card, handler)
client = TestClient(app_instance.build(agent_card_url='/my-agent'))
response = client.get('/my-agent')
assert response.status_code == 200
data = response.json()
assert data['name'] == agent_card.name
response = client.get(PREV_AGENT_CARD_WELL_KNOWN_PATH)
assert response.status_code == 404


def test_authenticated_extended_agent_card_endpoint_not_supported_fastapi(
agent_card: AgentCard, handler: mock.AsyncMock
):
Expand Down Expand Up @@ -253,9 +287,7 @@ def test_starlette_rpc_endpoint_custom_url(
"""Test the RPC endpoint with a custom URL."""
# Provide a valid Task object as the return value
task_status = TaskStatus(**MINIMAL_TASK_STATUS)
task = Task(
id='task1', context_id='ctx1', state='completed', status=task_status
)
task = Task(id='task1', context_id='ctx1', status=task_status)
handler.on_get_task.return_value = task
client = TestClient(app.build(rpc_url='/api/rpc'))
response = client.post(
Expand All @@ -278,9 +310,7 @@ def test_fastapi_rpc_endpoint_custom_url(
"""Test the RPC endpoint with a custom URL."""
# Provide a valid Task object as the return value
task_status = TaskStatus(**MINIMAL_TASK_STATUS)
task = Task(
id='task1', context_id='ctx1', state='completed', status=task_status
)
task = Task(id='task1', context_id='ctx1', status=task_status)
handler.on_get_task.return_value = task
client = TestClient(app.build(rpc_url='/api/rpc'))
response = client.post(
Expand Down Expand Up @@ -315,7 +345,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(AGENT_CARD_WELL_KNOWN_PATH)
assert response.status_code == 200
data = response.json()
assert data['name'] == agent_card.name
Expand All @@ -339,11 +369,40 @@ def custom_handler(request):
assert response.json() == {'message': 'Hello'}

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

# check if deprecated agent card path route is available with default well-known path
response = client.get(PREV_AGENT_CARD_WELL_KNOWN_PATH)
assert response.status_code == 200
data = response.json()
assert data['name'] == agent_card.name


def test_fastapi_build_custom_agent_card_path(
app: A2AFastAPIApplication, agent_card: AgentCard
):
"""Test building the app with a custom agent card path."""

test_app = app.build(agent_card_url='/agent-card')
client = TestClient(test_app)

# Ensure custom card path works
response = client.get('/agent-card')
assert response.status_code == 200
data = response.json()
assert data['name'] == agent_card.name

# Ensure default agent card location is not available
response = client.get(AGENT_CARD_WELL_KNOWN_PATH)
assert response.status_code == 404

# check if deprecated agent card path route is not available
response = client.get(PREV_AGENT_CARD_WELL_KNOWN_PATH)
assert response.status_code == 404


# === REQUEST METHODS TESTS ===

Expand Down Expand Up @@ -395,9 +454,7 @@ def test_cancel_task(client: TestClient, handler: mock.AsyncMock):
# Setup mock response
task_status = TaskStatus(**MINIMAL_TASK_STATUS)
task_status.state = TaskState.canceled # 'cancelled' #
task = Task(
id='task1', context_id='ctx1', state='cancelled', status=task_status
)
task = Task(id='task1', context_id='ctx1', status=task_status)
handler.on_cancel_task.return_value = task

# Send request
Expand Down Expand Up @@ -425,9 +482,7 @@ def test_get_task(client: TestClient, handler: mock.AsyncMock):
"""Test getting a task."""
# Setup mock response
task_status = TaskStatus(**MINIMAL_TASK_STATUS)
task = Task(
id='task1', context_id='ctx1', state='completed', status=task_status
)
task = Task(id='task1', context_id='ctx1', status=task_status)
handler.on_get_task.return_value = task # JSONRPCResponse(root=task)

# Send request
Expand Down
Loading