Adding support for proxy agent messages

Author: marktayl1

src/backend/v3/magentic_agents/magentic_agent_factory.py

@@ -9,6 +9,7 @@
 from typing import List, Union
 
 from common.models.messages_kernel import TeamConfiguration
+from v3.config.settings import current_user_id
 from v3.magentic_agents.foundry_agent import FoundryAgentTemplate
 from v3.magentic_agents.models.agent_models import (BingConfig, MCPConfig,
                                                     SearchConfig)
@@ -60,7 +61,8 @@ async def create_agent_from_config(self, agent_obj: SimpleNamespace) -> Union[Fo
 
         if not deployment_name and agent_obj.name.lower() == "proxyagent":
             self.logger.info("Creating ProxyAgent")
-            return ProxyAgent()
+            user_id = current_user_id.get()
+            return ProxyAgent(user_id=user_id)
 
         # Validate supported models
         supported_models = json.loads(os.getenv("SUPPORTED_MODELS"))

src/backend/v3/magentic_agents/proxy_agent.py

@@ -4,18 +4,23 @@
 import logging
 import uuid
 from collections.abc import AsyncIterable
-from typing import AsyncIterator
+from typing import AsyncIterator, Optional
 
+from pydantic import Field
 from semantic_kernel.agents import (  # pylint: disable=no-name-in-module
     AgentResponseItem, AgentThread)
 from semantic_kernel.agents.agent import Agent
-from semantic_kernel.contents import AuthorRole, ChatMessageContent
+from semantic_kernel.contents import (AuthorRole, ChatMessageContent,
+                                      StreamingChatMessageContent)
 from semantic_kernel.contents.chat_history import ChatHistory
 from semantic_kernel.contents.history_reducer.chat_history_reducer import \
     ChatHistoryReducer
 from semantic_kernel.exceptions.agent_exceptions import \
     AgentThreadOperationException
 from typing_extensions import override
+from v3.callbacks.response_handlers import (agent_response_callback,
+                                            streaming_agent_response_callback)
+from v3.config.settings import current_user_id
 
 
 class DummyAgentThread(AgentThread):
@@ -82,15 +87,48 @@ def __init__(self, message: ChatMessageContent, thread: AgentThread):
 
 class ProxyAgent(Agent):
     """Simple proxy agent that prompts for human clarification."""
+
+    # Declare as Pydantic field
+    user_id: Optional[str] = Field(default=None, description="User ID for WebSocket messaging")
 
-    def __init__(self):
+    def __init__(self, user_id: str = None, **kwargs):
+        # Get user_id from parameter or context, fallback to empty string
+        effective_user_id = user_id or current_user_id.get() or ""
         super().__init__(
-            name="ProxyAgent", 
+            name="ProxyAgent",
             description="""Call this agent when you need to clarify requests by asking the human user 
             for more information. Ask it for more details about any unclear requirements, missing information, 
-            or if you need the user to elaborate on any aspect of the task."""
+            or if you need the user to elaborate on any aspect of the task.""",
+            user_id=effective_user_id,
+            **kwargs
         )
         self.instructions = ""
+
+        self.user_id = user_id or current_user_id.get()
+
+    def _create_message_content(self, content: str, thread_id: str = None) -> ChatMessageContent:
+        """Create a ChatMessageContent with proper metadata."""
+        return ChatMessageContent(
+            role=AuthorRole.ASSISTANT,
+            content=content,
+            name=self.name,
+            metadata={"thread_id": thread_id} if thread_id else {}
+        )
+
+    async def _trigger_response_callbacks(self, message_content: ChatMessageContent):
+        """Manually trigger the same response callbacks used by other agents."""
+        # Trigger the standard agent response callback
+        agent_response_callback(message_content, self.user_id)
+
+    async def _trigger_streaming_callbacks(self, content: str, is_final: bool = False):
+        """Manually trigger streaming callbacks for real-time updates."""
+        streaming_message = StreamingChatMessageContent(
+            role=AuthorRole.ASSISTANT,
+            content=content,
+            name=self.name,
+            choice_index=0
+        )
+        await streaming_agent_response_callback(streaming_message, is_final, self.user_id)
 
     async def invoke(self, message: str,*, thread: AgentThread | None = None,**kwargs) -> AsyncIterator[ChatMessageContent]:
         """Ask human user for clarification about the message."""
@@ -101,10 +139,10 @@ async def invoke(self, message: str,*, thread: AgentThread | None = None,**kwarg
             construct_thread=lambda: DummyAgentThread(),
             expected_type=DummyAgentThread,
         )
-        # Replace with websocket call when available
-        print(f"\nProxyAgent: Another agent is asking for clarification about:")
-        print(f"   Request: {message}")
-        print("-" * 60)
+        # Send clarification request via response handlers
+        clarification_request = f"I need clarification about: {message}"
+        clarification_message = self._create_message_content(clarification_request, thread.id)
+        # await self._trigger_response_callbacks(clarification_message)
 
         # Get human input
         human_response = input("Please provide clarification: ").strip()
@@ -114,12 +152,11 @@ async def invoke(self, message: str,*, thread: AgentThread | None = None,**kwarg
 
         response = f"Human clarification: {human_response}"
 
-        chat_message = ChatMessageContent(
-            role=AuthorRole.ASSISTANT,
-            content=response,
-            name=self.name,
-            metadata={"thread_id": thread.id}
-        )
+        # Send response via response handlers
+        response_message = self._create_message_content(response, thread.id)
+        #await self._trigger_response_callbacks(response_message)
+
+        chat_message = response_message
 
         yield AgentResponseItem(
             message=chat_message,
@@ -144,10 +181,9 @@ async def invoke_stream(self, messages, thread=None, **kwargs) -> AsyncIterator[
         else:
             message = str(messages)
 
-        # Replace with websocket call when available
-        print(f"\nProxyAgent: Another agent is asking for clarification about:")
-        print(f"   Request: {message}")
-        print("-" * 60)
+        # Send clarification request via streaming callbacks
+        clarification_request = f"I need clarification about: {message}"
+        #await self._trigger_streaming_callbacks(clarification_request)
 
         # Get human input - replace with websocket call when available
         human_response = input("Please provide clarification: ").strip()
@@ -157,12 +193,10 @@ async def invoke_stream(self, messages, thread=None, **kwargs) -> AsyncIterator[
 
         response = f"Human clarification: {human_response}"
 
-        chat_message = ChatMessageContent(
-            role=AuthorRole.ASSISTANT,
-            content=response,
-            name=self.name,
-            metadata={"thread_id": thread.id}
-        )
+        # Send response via streaming callbacks
+        #await self._trigger_streaming_callbacks(response, is_final=True)
+
+        chat_message = self._create_message_content(response, thread.id)
 
         yield AgentResponseItem(
             message=chat_message,
@@ -184,6 +218,6 @@ async def get_response(self, chat_history, **kwargs):
             content="No clarification provided."
         )
 
-async def create_proxy_agent():
+async def create_proxy_agent(user_id: str = None):
     """Factory function for human proxy agent."""
-    return ProxyAgent()
+    return ProxyAgent(user_id=user_id)