Merge pull request #29 from syedfakher27/adk-middleware

Adk middleware

Author: syedfakher27

typescript-sdk/integrations/adk-middleware/src/adk_middleware/adk_agent.py

@@ -406,9 +406,10 @@ async def _handle_tool_result_submission(
         """
         thread_id = input.thread_id
 
-        # Extract tool results first 
+        # Extract tool results that is send by the frontend 
         tool_results = await self._extract_tool_results(input)
 
+        # if the tool results are not send by the fronted then call the tool function
         if not tool_results:
             logger.error(f"Tool result submission without tool results for thread {thread_id}")
             yield RunErrorEvent(
@@ -714,16 +715,25 @@ async def _start_background_execution(
         # Create dynamic toolset if tools provided and prepare tool updates
         toolset = None
         if input.tools:
-            toolset = ClientProxyToolset(
-                ag_ui_tools=input.tools,
-                event_queue=event_queue
-            )
 
             # Get existing tools from the agent
             existing_tools = []
             if hasattr(adk_agent, 'tools') and adk_agent.tools:
                 existing_tools = list(adk_agent.tools) if isinstance(adk_agent.tools, (list, tuple)) else [adk_agent.tools]
 
+            # if same tool is defined in frontend and backend then agent will only use the backend tool
+            input_tools = []
+            for input_tool in input.tools:
+                # Check if this input tool's name matches any existing tool
+                if not any(hasattr(existing_tool, '__name__') and input_tool.name == existing_tool.__name__ 
+                        for existing_tool in existing_tools):
+                    input_tools.append(input_tool)
+                        
+            toolset = ClientProxyToolset(
+                ag_ui_tools=input_tools,
+                event_queue=event_queue
+            )
+
             # Combine existing tools with our proxy toolset
             combined_tools = existing_tools + [toolset]
             agent_updates['tools'] = combined_tools
@@ -859,15 +869,16 @@ async def _run_adk_in_background(
                         logger.debug(f"Emitting event to queue: {type(ag_ui_event).__name__} (thread {input.thread_id}, queue size before: {event_queue.qsize()})")
                         await event_queue.put(ag_ui_event)
                         logger.debug(f"Event queued: {type(ag_ui_event).__name__} (thread {input.thread_id}, queue size after: {event_queue.qsize()})")
-                else:
-                    final_state = await self._session_manager.get_session_state(input.thread_id,app_name,user_id)
-                    ag_ui_event =  event_translator._create_state_snapshot_event(final_state)                    
-                    await event_queue.put(ag_ui_event)
+             
 
             # Force close any streaming messages
             async for ag_ui_event in event_translator.force_close_streaming_message():
                 await event_queue.put(ag_ui_event)
-            
+            # moving states snapshot events after the text event clousure to avoid this error https://github.com/Contextable/ag-ui/issues/28
+            final_state = await self._session_manager.get_session_state(input.thread_id,app_name,user_id)
+            if final_state:
+                ag_ui_event =  event_translator._create_state_snapshot_event(final_state)                    
+                await event_queue.put(ag_ui_event)
             # Signal completion - ADK execution is done
             logger.debug(f"Background task sending completion signal for thread {input.thread_id}")
             await event_queue.put(None)

typescript-sdk/integrations/adk-middleware/src/adk_middleware/event_translator.py

@@ -11,13 +11,13 @@
     BaseEvent, EventType,
     TextMessageStartEvent, TextMessageContentEvent, TextMessageEndEvent,
     ToolCallStartEvent, ToolCallArgsEvent, ToolCallEndEvent,
-    ToolCallChunkEvent,
+    ToolCallChunkEvent,ToolCallResultEvent,
     StateSnapshotEvent, StateDeltaEvent,
     MessagesSnapshotEvent,
     CustomEvent,
     Message, AssistantMessage, UserMessage, ToolMessage
 )
-
+import json
 from google.adk.events import Event as ADKEvent
 
 import logging
@@ -87,13 +87,6 @@ async def translate(
 
 
 
-            # Handle function responses
-            if hasattr(adk_event, 'get_function_responses'):
-                function_responses = adk_event.get_function_responses()
-                if function_responses:
-                    # Function responses are typically handled by the agent internally
-                    # We don't need to emit them as AG-UI events
-                    pass
 
             # call _translate_function_calls function to yield Tool Events
             if hasattr(adk_event, 'get_function_calls'):               
@@ -104,6 +97,15 @@ async def translate(
                     async for event in self._translate_function_calls(function_calls):
                         yield event
 
+            # Handle function responses and yield the tool response event
+            # this is essential for scenerios when user has to render function response at frontend
+            if hasattr(adk_event, 'get_function_responses'):
+                function_responses = adk_event.get_function_responses()
+                if function_responses:
+                    # Function responses should be emmitted to frontend so it can render the response as well
+                    async for event in self._translate_function_response(function_responses):
+                        yield event
+                    
 
             # Handle state changes
             if hasattr(adk_event, 'actions') and adk_event.actions and hasattr(adk_event.actions, 'state_delta') and adk_event.actions.state_delta:
@@ -281,6 +283,32 @@ async def _translate_function_calls(
             # Clean up tracking
             self._active_tool_calls.pop(tool_call_id, None)
 
+
+    async def _translate_function_response(
+        self,
+        function_response: list[types.FunctionResponse],
+    ) -> AsyncGenerator[BaseEvent, None]:
+        """Translate function calls from ADK event to AG-UI tool call events.
+        
+        Args:
+            adk_event: The ADK event containing function calls
+            function_response: List of function response from the event
+            
+        Yields:
+            Tool result events 
+        """
+        
+        for func_response in function_response:
+            
+            tool_call_id = getattr(func_response, 'id', str(uuid.uuid4()))
+            
+            yield ToolCallResultEvent(
+                message_id=str(uuid.uuid4()),
+                type=EventType.TOOL_CALL_RESULT,
+                tool_call_id=tool_call_id,
+                content=json.dumps(func_response.response)
+            )
+  
     def _create_state_delta_event(
         self,
         state_delta: Dict[str, Any],