lint, type, fmt fixes

Author: jackgerrits

python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_sequential_routed_agent.py

@@ -43,9 +43,9 @@ def __init__(self, description: str) -> None:
         super().__init__(description=description)
         self._fifo_lock = FIFOLock()
 
-    async def on_message(self, message: Any, ctx: MessageContext) -> Any | None:
+    async def on_message_impl(self, message: Any, ctx: MessageContext) -> None:
         await self._fifo_lock.acquire()
         try:
-            return await super().on_message(message, ctx)
+            await super().on_message_impl(message, ctx)
         finally:
             self._fifo_lock.release()

python/packages/autogen-core/docs/src/user-guide/core-user-guide/cookbook/local-llms-ollama-litellm.ipynb

@@ -174,7 +174,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 6,
+   "execution_count": null,
    "metadata": {},
    "outputs": [
     {
@@ -211,7 +211,6 @@
     "await runtime.send_message(\n",
     "    Message(\"Joe, tell me a joke.\"),\n",
     "    recipient=AgentId(joe, \"default\"),\n",
-    "    sender=AgentId(cathy, \"default\"),\n",
     ")\n",
     "await runtime.stop_when_idle()"
    ]

python/packages/autogen-core/src/autogen_core/application/_worker_runtime_host_servicer.py

@@ -102,18 +102,6 @@ async def _receive_messages(
             logger.info(f"Received message from client {client_id}: {message}")
             oneofcase = message.WhichOneof("message")
             match oneofcase:
-                case "request":
-                    request: agent_worker_pb2.RpcRequest = message.request
-                    task = asyncio.create_task(self._process_request(request, client_id))
-                    self._background_tasks.add(task)
-                    task.add_done_callback(self._raise_on_exception)
-                    task.add_done_callback(self._background_tasks.discard)
-                case "response":
-                    response: agent_worker_pb2.RpcResponse = message.response
-                    task = asyncio.create_task(self._process_response(response, client_id))
-                    self._background_tasks.add(task)
-                    task.add_done_callback(self._raise_on_exception)
-                    task.add_done_callback(self._background_tasks.discard)
                 case "cloudEvent":
                     # The proto typing doesnt resolve this one
                     event = cast(cloudevent_pb2.CloudEvent, message.cloudEvent)  # type: ignore
@@ -140,43 +128,6 @@ async def _receive_messages(
                 case None:
                     logger.warning("Received empty message")
 
-    async def _process_request(self, request: agent_worker_pb2.RpcRequest, client_id: int) -> None:
-        # Deliver the message to a client given the target agent type.
-        async with self._agent_type_to_client_id_lock:
-            target_client_id = self._agent_type_to_client_id.get(request.target.type)
-        if target_client_id is None:
-            logger.error(f"Agent {request.target.type} not found, failed to deliver message.")
-            return
-        target_send_queue = self._send_queues.get(target_client_id)
-        if target_send_queue is None:
-            logger.error(f"Client {target_client_id} not found, failed to deliver message.")
-            return
-        await target_send_queue.put(agent_worker_pb2.Message(request=request))
-
-        # Create a future to wait for the response from the target.
-        future = asyncio.get_event_loop().create_future()
-        self._pending_responses.setdefault(target_client_id, {})[request.request_id] = future
-
-        # Create a task to wait for the response and send it back to the client.
-        send_response_task = asyncio.create_task(self._wait_and_send_response(future, client_id))
-        self._background_tasks.add(send_response_task)
-        send_response_task.add_done_callback(self._raise_on_exception)
-        send_response_task.add_done_callback(self._background_tasks.discard)
-
-    async def _wait_and_send_response(self, future: Future[agent_worker_pb2.RpcResponse], client_id: int) -> None:
-        response = await future
-        message = agent_worker_pb2.Message(response=response)
-        send_queue = self._send_queues.get(client_id)
-        if send_queue is None:
-            logger.error(f"Client {client_id} not found, failed to send response message.")
-            return
-        await send_queue.put(message)
-
-    async def _process_response(self, response: agent_worker_pb2.RpcResponse, client_id: int) -> None:
-        # Setting the result of the future will send the response back to the original sender.
-        future = self._pending_responses[client_id].pop(response.request_id)
-        future.set_result(response)
-
     async def _process_event(self, event: cloudevent_pb2.CloudEvent) -> None:
         topic_id = TopicId(type=event.type, source=event.source)
         recipients = await self._subscription_manager.get_subscribed_recipients(topic_id)

python/packages/autogen-core/src/autogen_core/base/_agent_proxy.py

@@ -29,13 +29,11 @@ async def send_message(
         self,
         message: Any,
         *,
-        sender: AgentId,
         cancellation_token: CancellationToken | None = None,
     ) -> Any:
         return await self._runtime.send_message(
             message,
             recipient=self._agent,
-            sender=sender,
             cancellation_token=cancellation_token,
         )
 

python/packages/autogen-core/src/autogen_core/base/_base_agent.py

@@ -126,7 +126,7 @@ async def on_message_impl(self, message: Any, ctx: MessageContext) -> None: ...
     @final
     async def on_message(self, message: Any, ctx: MessageContext) -> None:
         # Intercept RPC responses
-        if ctx.topic_id is not None and (request_id := is_rpc_response(ctx.topic_id.type)) is not None:
+        if (request_id := is_rpc_response(ctx.topic_id.type)) is not None:
             if request_id in self._pending_rpc_requests:
                 self._pending_rpc_requests[request_id].set_result(message)
                 del self._pending_rpc_requests[request_id]

python/packages/autogen-core/src/autogen_core/components/_routed_agent.py

@@ -515,7 +515,7 @@ def __init__(self, description: str) -> None:
 
         super().__init__(description)
 
-    async def on_message_impl(self, message: Any, ctx: MessageContext):
+    async def on_message_impl(self, message: Any, ctx: MessageContext) -> None:
         """Handle a message by routing it to the appropriate message handler.
         Do not override this method in subclasses. Instead, add message handlers as methods decorated with
         either the :func:`event` or :func:`rpc` decorator."""
@@ -526,8 +526,8 @@ async def on_message_impl(self, message: Any, ctx: MessageContext):
             # Call the first handler whose router returns True and then return the result.
             for h in handlers:
                 if h.router(message, ctx):
-                    return await h(self, message, ctx)
-        return await self.on_unhandled_message(message, ctx)  # type: ignore
+                    await h(self, message, ctx)
+        await self.on_unhandled_message(message, ctx)
 
     async def on_unhandled_message(self, message: Any, ctx: MessageContext) -> None:
         """Called when a message is received that does not have a matching message handler.

python/packages/autogen-core/tests/test_routed_agent.py

@@ -5,6 +5,7 @@
 import pytest
 from autogen_core.application import SingleThreadedAgentRuntime
 from autogen_core.base import AgentId, MessageContext, TopicId
+from autogen_core.base._rpc import is_rpc_request
 from autogen_core.components import RoutedAgent, TypeSubscription, event, message_handler, rpc
 from test_utils import LoopbackAgent
 
@@ -23,12 +24,12 @@ def __init__(self) -> None:
         self.num_calls_rpc = 0
         self.num_calls_broadcast = 0
 
-    @message_handler(match=lambda _, ctx: ctx.is_rpc)
+    @message_handler(match=lambda _, ctx: is_rpc_request(ctx.topic_id.type) is not None)
     async def on_rpc_message(self, message: MessageType, ctx: MessageContext) -> MessageType:
         self.num_calls_rpc += 1
         return message
 
-    @message_handler(match=lambda _, ctx: not ctx.is_rpc)
+    @message_handler(match=lambda _, ctx: is_rpc_request(ctx.topic_id.type) is None)
     async def on_broadcast_message(self, message: MessageType, ctx: MessageContext) -> None:
         self.num_calls_broadcast += 1
 

python/packages/autogen-magentic-one/tests/headless_web_surfer/test_web_surfer.py

@@ -218,27 +218,22 @@ async def test_web_surfer_oai() -> None:
             )
         ),
         recipient=web_surfer.id,
-        sender=user_proxy.id,
     )
     await runtime.send_message(
         BroadcastMessage(content=UserMessage(content="Please scroll down.", source="user")),
         recipient=web_surfer.id,
-        sender=user_proxy.id,
     )
     await runtime.send_message(
         BroadcastMessage(content=UserMessage(content="Please scroll up.", source="user")),
         recipient=web_surfer.id,
-        sender=user_proxy.id,
     )
     await runtime.send_message(
         BroadcastMessage(content=UserMessage(content="When was it founded?", source="user")),
         recipient=web_surfer.id,
-        sender=user_proxy.id,
     )
     await runtime.send_message(
         BroadcastMessage(content=UserMessage(content="What's this page about?", source="user")),
         recipient=web_surfer.id,
-        sender=user_proxy.id,
     )
     await runtime.stop_when_idle()