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
80 changes: 79 additions & 1 deletion src/a2a/types.py
Original file line number Diff line number Diff line change
Expand Up @@ -172,6 +172,26 @@ class AgentSkill(A2ABaseModel):
"""


class AuthenticatedExtendedCardNotConfiguredError(A2ABaseModel):
"""
An A2A-specific error indicating that the agent does not have an Authenticated Extended Card configured
"""

code: Literal[-32007] = -32007
"""
The error code for when an authenticated extended card is not configured.
"""
data: Any | None = None
"""
A primitive or structured value containing additional information about the error.
This may be omitted.
"""
message: str | None = 'Authenticated Extended Card is not configured'
"""
The error message.
"""


class AuthorizationCodeOAuthFlow(A2ABaseModel):
"""
Defines configuration details for the OAuth 2.0 Authorization Code flow.
Expand Down Expand Up @@ -375,6 +395,27 @@ class FileWithUri(A2ABaseModel):
"""


class GetAuthenticatedExtendedCardRequest(A2ABaseModel):
"""
Represents a JSON-RPC request for the `agent/getAuthenticatedExtendedCard` method.
"""

id: str | int
"""
The identifier for this request.
"""
jsonrpc: Literal['2.0'] = '2.0'
"""
The version of the JSON-RPC protocol. MUST be exactly "2.0".
"""
method: Literal['agent/getAuthenticatedExtendedCard'] = (
'agent/getAuthenticatedExtendedCard'
)
"""
The method name. Must be 'agent/getAuthenticatedExtendedCard'.
"""


class GetTaskPushNotificationConfigParams(A2ABaseModel):
"""
Defines parameters for fetching a specific push notification configuration for a task.
Expand Down Expand Up @@ -999,6 +1040,7 @@ class A2AError(
| UnsupportedOperationError
| ContentTypeNotSupportedError
| InvalidAgentResponseError
| AuthenticatedExtendedCardNotConfiguredError
]
):
root: (
Expand All @@ -1013,6 +1055,7 @@ class A2AError(
| UnsupportedOperationError
| ContentTypeNotSupportedError
| InvalidAgentResponseError
| AuthenticatedExtendedCardNotConfiguredError
)
"""
A discriminated union of all standard JSON-RPC and A2A-specific error types.
Expand Down Expand Up @@ -1170,6 +1213,7 @@ class JSONRPCErrorResponse(A2ABaseModel):
| UnsupportedOperationError
| ContentTypeNotSupportedError
| InvalidAgentResponseError
| AuthenticatedExtendedCardNotConfiguredError
)
"""
An object describing the error that occurred.
Expand Down Expand Up @@ -1625,6 +1669,7 @@ class A2ARequest(
| TaskResubscriptionRequest
| ListTaskPushNotificationConfigRequest
| DeleteTaskPushNotificationConfigRequest
| GetAuthenticatedExtendedCardRequest
]
):
root: (
Expand All @@ -1637,6 +1682,7 @@ class A2ARequest(
| TaskResubscriptionRequest
| ListTaskPushNotificationConfigRequest
| DeleteTaskPushNotificationConfigRequest
| GetAuthenticatedExtendedCardRequest
)
"""
A discriminated union representing all possible JSON-RPC 2.0 requests supported by the A2A specification.
Expand Down Expand Up @@ -1750,6 +1796,25 @@ class AgentCard(A2ABaseModel):
"""


class GetAuthenticatedExtendedCardSuccessResponse(A2ABaseModel):
"""
Represents a successful JSON-RPC response for the `agent/getAuthenticatedExtendedCard` method.
"""

id: str | int | None = None
"""
The identifier established by the client.
"""
jsonrpc: Literal['2.0'] = '2.0'
"""
The version of the JSON-RPC protocol. MUST be exactly "2.0".
"""
result: AgentCard
"""
The result is an Agent Card object.
"""


class Task(A2ABaseModel):
"""
Represents a single, stateful operation or conversation between a client and an agent.
Expand All @@ -1769,7 +1834,7 @@ class Task(A2ABaseModel):
"""
id: str
"""
A unique identifier for the task, generated by the client for a new task or provided by the agent.
A unique identifier for the task, generated by the server for a new task.
"""
kind: Literal['task'] = 'task'
"""
Expand Down Expand Up @@ -1804,6 +1869,17 @@ class CancelTaskSuccessResponse(A2ABaseModel):
"""


class GetAuthenticatedExtendedCardResponse(
RootModel[
JSONRPCErrorResponse | GetAuthenticatedExtendedCardSuccessResponse
]
):
root: JSONRPCErrorResponse | GetAuthenticatedExtendedCardSuccessResponse
"""
Represents a JSON-RPC response for the `agent/getAuthenticatedExtendedCard` method.
"""


class GetTaskSuccessResponse(A2ABaseModel):
"""
Represents a successful JSON-RPC response for the `tasks/get` method.
Expand Down Expand Up @@ -1889,6 +1965,7 @@ class JSONRPCResponse(
| GetTaskPushNotificationConfigSuccessResponse
| ListTaskPushNotificationConfigSuccessResponse
| DeleteTaskPushNotificationConfigSuccessResponse
| GetAuthenticatedExtendedCardSuccessResponse
]
):
root: (
Expand All @@ -1901,6 +1978,7 @@ class JSONRPCResponse(
| GetTaskPushNotificationConfigSuccessResponse
| ListTaskPushNotificationConfigSuccessResponse
| DeleteTaskPushNotificationConfigSuccessResponse
| GetAuthenticatedExtendedCardSuccessResponse
)
"""
A discriminated union representing all possible JSON-RPC 2.0 responses
Expand Down
3 changes: 2 additions & 1 deletion tests/server/test_integration.py
Original file line number Diff line number Diff line change
Expand Up @@ -836,7 +836,8 @@ def test_invalid_request_structure(client: TestClient):
'/',
json={
# Missing required fields
'id': '123'
'id': '123',
'method': 'foo/bar',
},
)
assert response.status_code == 200
Expand Down
85 changes: 82 additions & 3 deletions tests/test_types.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,9 @@
FileWithUri,
GetTaskPushNotificationConfigParams,
GetTaskPushNotificationConfigRequest,
GetAuthenticatedExtendedCardRequest,
GetAuthenticatedExtendedCardResponse,
GetAuthenticatedExtendedCardSuccessResponse,
GetTaskPushNotificationConfigResponse,
GetTaskPushNotificationConfigSuccessResponse,
GetTaskRequest,
Expand Down Expand Up @@ -968,6 +971,21 @@ def test_a2a_request_root_model() -> None:
assert isinstance(a2a_req_task_resubscribe_req.root.params, TaskIdParams)
assert a2a_req_task_resubscribe_req.root.method == 'tasks/resubscribe'

# GetAuthenticatedExtendedCardRequest
get_auth_card_req_data: dict[str, Any] = {
'jsonrpc': '2.0',
'method': 'agent/getAuthenticatedExtendedCard',
'id': 2,
}
a2a_req_get_auth_card = A2ARequest.model_validate(get_auth_card_req_data)
assert isinstance(
a2a_req_get_auth_card.root, GetAuthenticatedExtendedCardRequest
)
assert (
a2a_req_get_auth_card.root.method
== 'agent/getAuthenticatedExtendedCard'
)

# Invalid method case
invalid_req_data: dict[str, Any] = {
'jsonrpc': '2.0',
Expand Down Expand Up @@ -1055,6 +1073,14 @@ def test_a2a_request_root_model_id_validation() -> None:
with pytest.raises(ValidationError):
A2ARequest.model_validate(task_resubscribe_req_data)

# GetAuthenticatedExtendedCardRequest
get_auth_card_req_data: dict[str, Any] = {
'jsonrpc': '2.0',
'method': 'agent/getAuthenticatedExtendedCard',
}
with pytest.raises(ValidationError):
A2ARequest.model_validate(get_auth_card_req_data) # missing id


def test_content_type_not_supported_error():
# Test ContentTypeNotSupportedError
Expand Down Expand Up @@ -1544,11 +1570,11 @@ def test_camelCase() -> None:
description='Just a hello world agent',
url='http://localhost:9999/',
version='1.0.0',
defaultInputModes=['text'],
defaultOutputModes=['text'],
defaultInputModes=['text'], # type: ignore
defaultOutputModes=['text'], # type: ignore
capabilities=AgentCapabilities(streaming=True),
skills=[skill],
supportsAuthenticatedExtendedCard=True,
supportsAuthenticatedExtendedCard=True, # type: ignore
)

# Test setting an attribute via camelCase alias
Expand All @@ -1568,3 +1594,56 @@ def test_camelCase() -> None:
assert agent_card.supports_authenticated_extended_card is False
assert default_input_modes == ['text']
assert agent_card.default_input_modes == ['text']


def test_get_authenticated_extended_card_request() -> None:
req_data: dict[str, Any] = {
'jsonrpc': '2.0',
'method': 'agent/getAuthenticatedExtendedCard',
'id': 5,
}
req = GetAuthenticatedExtendedCardRequest.model_validate(req_data)
assert req.method == 'agent/getAuthenticatedExtendedCard'
assert req.id == 5
# This request has no params, so we don't check for that.

with pytest.raises(ValidationError): # Wrong method literal
GetAuthenticatedExtendedCardRequest.model_validate(
{**req_data, 'method': 'wrong/method'}
)

with pytest.raises(ValidationError): # Missing id
GetAuthenticatedExtendedCardRequest.model_validate(
{'jsonrpc': '2.0', 'method': 'agent/getAuthenticatedExtendedCard'}
)


def test_get_authenticated_extended_card_response() -> None:
resp_data: dict[str, Any] = {
'jsonrpc': '2.0',
'result': MINIMAL_AGENT_CARD,
'id': 'resp-1',
}
resp = GetAuthenticatedExtendedCardResponse.model_validate(resp_data)
assert resp.root.id == 'resp-1'
assert isinstance(resp.root, GetAuthenticatedExtendedCardSuccessResponse)
assert isinstance(resp.root.result, AgentCard)
assert resp.root.result.name == 'TestAgent'

with pytest.raises(ValidationError): # Result is not an AgentCard
GetAuthenticatedExtendedCardResponse.model_validate(
{'jsonrpc': '2.0', 'result': {'wrong': 'data'}, 'id': 1}
)

resp_data_err: dict[str, Any] = {
'jsonrpc': '2.0',
'error': JSONRPCError(**TaskNotFoundError().model_dump()),
'id': 'resp-1',
}
resp_err = GetAuthenticatedExtendedCardResponse.model_validate(
resp_data_err
)
assert resp_err.root.id == 'resp-1'
assert isinstance(resp_err.root, JSONRPCErrorResponse)
assert resp_err.root.error is not None
assert isinstance(resp_err.root.error, JSONRPCError)
Loading