wip

Author: ishymko

src/a2a/client/transports/grpc.py

@@ -76,7 +76,7 @@ def _map_grpc_error(e: grpc.aio.AioRpcError) -> NoReturn:
                 data = {'errors': errors}
                 exception_cls = InvalidParamsError
                 break
-            elif detail.Is(error_details_pb2.ErrorInfo.DESCRIPTOR):
+            if detail.Is(error_details_pb2.ErrorInfo.DESCRIPTOR):
                 error_info = error_details_pb2.ErrorInfo()
                 detail.Unpack(error_info)
                 if error_info.domain == 'a2a-protocol.org':

src/a2a/server/apps/rest/rest_adapter.py

@@ -159,7 +159,7 @@ async def event_generator(
                 yield json.dumps(item)
 
         return EventSourceResponse(
-            event_generator(method(request, call_context))
+            event_generator(await method(request, call_context))
         )
 
     async def handle_get_agent_card(

src/a2a/server/request_handlers/grpc_handler.py

@@ -406,7 +406,11 @@ async def abort_context(
             # Create standard Status
             status = status_pb2.Status(code=status_code, message=error_msg)
 
-            if isinstance(error, types.InvalidParamsError) and error.data and error.data.get('errors'):
+            if (
+                isinstance(error, types.InvalidParamsError)
+                and error.data
+                and error.data.get('errors')
+            ):
                 bad_request = error_details_pb2.BadRequest()
                 for err_dict in error.data['errors']:
                     violation = bad_request.field_violations.add()

src/a2a/server/request_handlers/request_handler.py

@@ -229,25 +229,24 @@ async def on_delete_task_push_notification_config(
 
 def validate_request_params(method: Callable) -> Callable:
     """Decorator for RequestHandler methods to validate required fields on incoming requests."""
-    if inspect.isasyncgenfunction(method):
+    if inspect.iscoroutinefunction(method):
 
         @functools.wraps(method)
-        async def async_generator_wrapper(
+        async def async_wrapper(
             self: RequestHandler,
             params: ProtoMessage,
             context: ServerCallContext,
             *args: Any,
             **kwargs: Any,
-        ) -> AsyncGenerator:
+        ) -> Any:
             if params is not None:
                 validate_proto_required_fields(params)
-            async for item in method(self, params, context, *args, **kwargs):
-                yield item
+            return await method(self, params, context, *args, **kwargs)
 
-        return async_generator_wrapper
+        return async_wrapper
 
     @functools.wraps(method)
-    async def async_wrapper(
+    def sync_wrapper(
         self: RequestHandler,
         params: ProtoMessage,
         context: ServerCallContext,
@@ -256,6 +255,6 @@ async def async_wrapper(
     ) -> Any:
         if params is not None:
             validate_proto_required_fields(params)
-        return await method(self, params, context, *args, **kwargs)
+        return method(self, params, context, *args, **kwargs)
 
-    return async_wrapper
+    return sync_wrapper

src/a2a/server/request_handlers/rest_handler.py

@@ -116,11 +116,14 @@ async def on_message_send_stream(
         body = await request.body()
         params = a2a_pb2.SendMessageRequest()
         Parse(body, params)
-        async for event in self.request_handler.on_message_send_stream(
-            params, context
-        ):
-            response = proto_utils.to_stream_response(event)
-            yield MessageToDict(response)
+        stream = self.request_handler.on_message_send_stream(params, context)
+
+        async def _generator() -> AsyncIterator[dict[str, Any]]:
+            async for event in stream:
+                response = proto_utils.to_stream_response(event)
+                yield MessageToDict(response)
+
+        return _generator()
 
     @validate_version(constants.PROTOCOL_VERSION_1_0)
     async def on_cancel_task(
@@ -167,10 +170,15 @@ async def on_subscribe_to_task(
             JSON serialized objects containing streaming events
         """
         task_id = request.path_params['id']
-        async for event in self.request_handler.on_subscribe_to_task(
+        stream = self.request_handler.on_subscribe_to_task(
             SubscribeToTaskRequest(id=task_id), context
-        ):
-            yield MessageToDict(proto_utils.to_stream_response(event))
+        )
+
+        async def _generator() -> AsyncIterator[dict[str, Any]]:
+            async for event in stream:
+                yield MessageToDict(proto_utils.to_stream_response(event))
+
+        return _generator()
 
     @validate_version(constants.PROTOCOL_VERSION_1_0)
     async def get_push_notification(