Merge branch 'main' into fix/streaming-endpoint-deadlock

Author: youngchannelforyou

.gitignore

@@ -6,6 +6,7 @@ __pycache__
 .pytest_cache
 .ruff_cache
 .venv
+test_venv/
 coverage.xml
 .nox
-spec.json
+spec.json

.ruff.toml

@@ -30,7 +30,6 @@ ignore = [
     "ANN003",
     "ANN401",
     "TRY003",
-    "G004",
     "TRY201",
     "FIX002",
 ]

src/a2a/client/auth/interceptor.py

@@ -62,7 +62,9 @@ async def intercept(
                         ):
                             headers['Authorization'] = f'Bearer {credential}'
                             logger.debug(
-                                f"Added Bearer token for scheme '{scheme_name}' (type: {scheme_def.type})."
+                                "Added Bearer token for scheme '%s' (type: %s).",
+                                scheme_name,
+                                scheme_def.type,
                             )
                             http_kwargs['headers'] = headers
                             return request_payload, http_kwargs
@@ -74,7 +76,9 @@ async def intercept(
                         ):
                             headers['Authorization'] = f'Bearer {credential}'
                             logger.debug(
-                                f"Added Bearer token for scheme '{scheme_name}' (type: {scheme_def.type})."
+                                "Added Bearer token for scheme '%s' (type: %s).",
+                                scheme_name,
+                                scheme_def.type,
                             )
                             http_kwargs['headers'] = headers
                             return request_payload, http_kwargs
@@ -83,7 +87,8 @@ async def intercept(
                         case APIKeySecurityScheme(in_=In.header):
                             headers[scheme_def.name] = credential
                             logger.debug(
-                                f"Added API Key Header for scheme '{scheme_name}'."
+                                "Added API Key Header for scheme '%s'.",
+                                scheme_name,
                             )
                             http_kwargs['headers'] = headers
                             return request_payload, http_kwargs

src/a2a/server/apps/jsonrpc/jsonrpc_app.py

@@ -28,13 +28,15 @@
     GetTaskPushNotificationConfigRequest,
     GetTaskRequest,
     InternalError,
+    InvalidParamsError,
     InvalidRequestError,
     JSONParseError,
     JSONRPCError,
     JSONRPCErrorResponse,
     JSONRPCRequest,
     JSONRPCResponse,
     ListTaskPushNotificationConfigRequest,
+    MethodNotFoundError,
     SendMessageRequest,
     SendStreamingMessageRequest,
     SendStreamingMessageResponse,
@@ -89,6 +91,8 @@
         Response = Any
         HTTP_413_REQUEST_ENTITY_TOO_LARGE = Any
 
+MAX_CONTENT_LENGTH = 1_000_000
+
 
 class StarletteUserProxy(A2AUser):
     """Adapts the Starlette User class to the A2A user representation."""
@@ -151,6 +155,25 @@ class JSONRPCApplication(ABC):
     (SSE).
     """
 
+    # Method-to-model mapping for centralized routing
+    A2ARequestModel = (
+        SendMessageRequest
+        | SendStreamingMessageRequest
+        | GetTaskRequest
+        | CancelTaskRequest
+        | SetTaskPushNotificationConfigRequest
+        | GetTaskPushNotificationConfigRequest
+        | ListTaskPushNotificationConfigRequest
+        | DeleteTaskPushNotificationConfigRequest
+        | TaskResubscriptionRequest
+        | GetAuthenticatedExtendedCardRequest
+    )
+
+    METHOD_TO_MODEL: dict[str, type[A2ARequestModel]] = {
+        model.model_fields['method'].default: model
+        for model in A2ARequestModel.__args__
+    }
+
     def __init__(  # noqa: PLR0913
         self,
         agent_card: AgentCard,
@@ -233,9 +256,13 @@ def _generate_error_response(
         )
         logger.log(
             log_level,
-            f'Request Error (ID: {request_id}): '
-            f"Code={error_resp.error.code}, Message='{error_resp.error.message}'"
-            f'{", Data=" + str(error_resp.error.data) if error_resp.error.data else ""}',
+            "Request Error (ID: %s): Code=%s, Message='%s'%s",
+            request_id,
+            error_resp.error.code,
+            error_resp.error.message,
+            ', Data=' + str(error_resp.error.data)
+            if error_resp.error.data
+            else '',
         )
         return JSONResponse(
             error_resp.model_dump(mode='json', exclude_none=True),
@@ -267,17 +294,60 @@ async def _handle_requests(self, request: Request) -> Response:  # noqa: PLR0911
             body = await request.json()
             if isinstance(body, dict):
                 request_id = body.get('id')
+                # Ensure request_id is valid for JSON-RPC response (str/int/None only)
+                if request_id is not None and not isinstance(
+                    request_id, str | int
+                ):
+                    request_id = None
+            # Treat very large payloads as invalid request (-32600) before routing
+            with contextlib.suppress(Exception):
+                content_length = int(request.headers.get('content-length', '0'))
+                if content_length and content_length > MAX_CONTENT_LENGTH:
+                    return self._generate_error_response(
+                        request_id,
+                        A2AError(
+                            root=InvalidRequestError(
+                                message='Payload too large'
+                            )
+                        ),
+                    )
+            logger.debug('Request body: %s', body)
+            # 1) Validate base JSON-RPC structure only (-32600 on failure)
+            try:
+                base_request = JSONRPCRequest.model_validate(body)
+            except ValidationError as e:
+                logger.exception('Failed to validate base JSON-RPC request')
+                return self._generate_error_response(
+                    request_id,
+                    A2AError(
+                        root=InvalidRequestError(data=json.loads(e.json()))
+                    ),
+                )
 
-            # First, validate the basic JSON-RPC structure. This is crucial
-            # because the A2ARequest model is a discriminated union where some
-            # request types have default values for the 'method' field
-            JSONRPCRequest.model_validate(body)
+            # 2) Route by method name; unknown -> -32601, known -> validate params (-32602 on failure)
+            method = base_request.method
 
-            a2a_request = A2ARequest.model_validate(body)
+            model_class = self.METHOD_TO_MODEL.get(method)
+            if not model_class:
+                return self._generate_error_response(
+                    request_id, A2AError(root=MethodNotFoundError())
+                )
+            try:
+                specific_request = model_class.model_validate(body)
+            except ValidationError as e:
+                logger.exception('Failed to validate base JSON-RPC request')
+                return self._generate_error_response(
+                    request_id,
+                    A2AError(
+                        root=InvalidParamsError(data=json.loads(e.json()))
+                    ),
+                )
 
+            # 3) Build call context and wrap the request for downstream handling
             call_context = self._context_builder.build(request)
 
-            request_id = a2a_request.root.id
+            request_id = specific_request.id
+            a2a_request = A2ARequest(root=specific_request)
             request_obj = a2a_request.root
 
             if isinstance(
@@ -301,12 +371,6 @@ async def _handle_requests(self, request: Request) -> Response:  # noqa: PLR0911
             return self._generate_error_response(
                 None, A2AError(root=JSONParseError(message=str(e)))
             )
-        except ValidationError as e:
-            traceback.print_exc()
-            return self._generate_error_response(
-                request_id,
-                A2AError(root=InvalidRequestError(data=json.loads(e.json()))),
-            )
         except HTTPException as e:
             if e.status_code == HTTP_413_REQUEST_ENTITY_TOO_LARGE:
                 return self._generate_error_response(
@@ -422,7 +486,7 @@ async def _process_non_streaming_request(
                 )
             case _:
                 logger.error(
-                    f'Unhandled validated request type: {type(request_obj)}'
+                    'Unhandled validated request type: %s', type(request_obj)
                 )
                 error = UnsupportedOperationError(
                     message=f'Request type {type(request_obj).__name__} is unknown.'
@@ -497,8 +561,10 @@ async def _handle_get_agent_card(self, request: Request) -> JSONResponse:
         """
         if request.url.path == PREV_AGENT_CARD_WELL_KNOWN_PATH:
             logger.warning(
-                f"Deprecated agent card endpoint '{PREV_AGENT_CARD_WELL_KNOWN_PATH}' accessed. "
-                f"Please use '{AGENT_CARD_WELL_KNOWN_PATH}' instead. This endpoint will be removed in a future version."
+                "Deprecated agent card endpoint '%s' accessed. "
+                "Please use '%s' instead. This endpoint will be removed in a future version.",
+                PREV_AGENT_CARD_WELL_KNOWN_PATH,
+                AGENT_CARD_WELL_KNOWN_PATH,
             )
 
         card_to_serve = self.agent_card

src/a2a/server/events/event_consumer.py

@@ -62,7 +62,7 @@ async def consume_one(self) -> Event:
                 InternalError(message='Agent did not return any response')
             ) from e
 
-        logger.debug(f'Dequeued event of type: {type(event)} in consume_one.')
+        logger.debug('Dequeued event of type: %s in consume_one.', type(event))
 
         self.queue.task_done()
 
@@ -95,7 +95,7 @@ async def consume_all(self) -> AsyncGenerator[Event]:
                     self.queue.dequeue_event(), timeout=self._timeout
                 )
                 logger.debug(
-                    f'Dequeued event of type: {type(event)} in consume_all.'
+                    'Dequeued event of type: %s in consume_all.', type(event)
                 )
                 self.queue.task_done()
                 logger.debug(

src/a2a/server/events/event_queue.py

@@ -54,7 +54,7 @@ async def enqueue_event(self, event: Event) -> None:
                 logger.warning('Queue is closed. Event will not be enqueued.')
                 return
 
-        logger.debug(f'Enqueuing event of type: {type(event)}')
+        logger.debug('Enqueuing event of type: %s', type(event))
 
         # Make sure to use put instead of put_nowait to avoid blocking the event loop.
         await self.queue.put(event)
@@ -103,13 +103,13 @@ async def dequeue_event(self, no_wait: bool = False) -> Event:
             logger.debug('Attempting to dequeue event (no_wait=True).')
             event = self.queue.get_nowait()
             logger.debug(
-                f'Dequeued event (no_wait=True) of type: {type(event)}'
+                'Dequeued event (no_wait=True) of type: %s', type(event)
             )
             return event
 
         logger.debug('Attempting to dequeue event (waiting).')
         event = await self.queue.get()
-        logger.debug(f'Dequeued event (waited) of type: {type(event)}')
+        logger.debug('Dequeued event (waited) of type: %s', type(event))
         return event
 
     def task_done(self) -> None:
@@ -193,7 +193,9 @@ async def clear_events(self, clear_child_queues: bool = True) -> None:
                 while True:
                     event = self.queue.get_nowait()
                     logger.debug(
-                        f'Discarding unprocessed event of type: {type(event)}, content: {event}'
+                        'Discarding unprocessed event of type: %s, content: %s',
+                        type(event),
+                        event,
                     )
                     self.queue.task_done()
                     cleared_count += 1
@@ -211,7 +213,8 @@ async def clear_events(self, clear_child_queues: bool = True) -> None:
 
             if cleared_count > 0:
                 logger.debug(
-                    f'Cleared {cleared_count} unprocessed events from EventQueue.'
+                    'Cleared %d unprocessed events from EventQueue.',
+                    cleared_count,
                 )
 
         # Clear all child queues (lock released before awaiting child tasks)

src/a2a/server/models.py

@@ -147,14 +147,9 @@ def task_metadata(cls) -> Mapped[dict[str, Any] | None]:
     @override
     def __repr__(self) -> str:
         """Return a string representation of the task."""
-        repr_template = (
-            '<{CLS}(id="{ID}", context_id="{CTX_ID}", status="{STATUS}")>'
-        )
-        return repr_template.format(
-            CLS=self.__class__.__name__,
-            ID=self.id,
-            CTX_ID=self.context_id,
-            STATUS=self.status,
+        return (
+            f'<{self.__class__.__name__}(id="{self.id}", '
+            f'context_id="{self.context_id}", status="{self.status}")>'
         )
 
 
@@ -188,12 +183,9 @@ class TaskModel(TaskMixin, base):  # type: ignore
         @override
         def __repr__(self) -> str:
             """Return a string representation of the task."""
-            repr_template = '<TaskModel[{TABLE}](id="{ID}", context_id="{CTX_ID}", status="{STATUS}")>'
-            return repr_template.format(
-                TABLE=table_name,
-                ID=self.id,
-                CTX_ID=self.context_id,
-                STATUS=self.status,
+            return (
+                f'<TaskModel[{table_name}](id="{self.id}", '
+                f'context_id="{self.context_id}", status="{self.status}")>'
             )
 
     # Set a dynamic name for better debugging
@@ -221,11 +213,9 @@ class PushNotificationConfigMixin:
     @override
     def __repr__(self) -> str:
         """Return a string representation of the push notification config."""
-        repr_template = '<{CLS}(task_id="{TID}", config_id="{CID}")>'
-        return repr_template.format(
-            CLS=self.__class__.__name__,
-            TID=self.task_id,
-            CID=self.config_id,
+        return (
+            f'<{self.__class__.__name__}(task_id="{self.task_id}", '
+            f'config_id="{self.config_id}")>'
         )
 
 
@@ -241,11 +231,9 @@ class PushNotificationConfigModel(PushNotificationConfigMixin, base):  # type: i
         @override
         def __repr__(self) -> str:
             """Return a string representation of the push notification config."""
-            repr_template = '<PushNotificationConfigModel[{TABLE}](task_id="{TID}", config_id="{CID}")>'
-            return repr_template.format(
-                TABLE=table_name,
-                TID=self.task_id,
-                CID=self.config_id,
+            return (
+                f'<PushNotificationConfigModel[{table_name}]('
+                f'task_id="{self.task_id}", config_id="{self.config_id}")>'
             )
 
     PushNotificationConfigModel.__name__ = (