Merge pull request #155 from scaleapi/dm/streaming-bug-fix

Fix streaming bugs sync

Author: danielmillerp

src/agentex/lib/adk/providers/_modules/openai.py

@@ -65,9 +65,9 @@ def __init__(
     async def run_agent(
         self,
         input_list: list[dict[str, Any]],
-        mcp_server_params: list[StdioServerParameters],
         agent_name: str,
         agent_instructions: str,
+        mcp_server_params: list[StdioServerParameters] | None = None,
         trace_id: str | None = None,
         parent_span_id: str | None = None,
         start_to_close_timeout: timedelta = timedelta(seconds=600),
@@ -119,6 +119,10 @@ async def run_agent(
         Returns:
             Union[SerializableRunResult, RunResult]: SerializableRunResult when in Temporal, RunResult otherwise.
         """
+        # Default to empty list if not provided
+        if mcp_server_params is None:
+            mcp_server_params = []
+
         if in_temporal_workflow():
             params = RunAgentParams(
                 trace_id=trace_id,
@@ -174,9 +178,9 @@ async def run_agent_auto_send(
         self,
         task_id: str,
         input_list: list[dict[str, Any]],
-        mcp_server_params: list[StdioServerParameters],
         agent_name: str,
         agent_instructions: str,
+        mcp_server_params: list[StdioServerParameters] | None = None,
         trace_id: str | None = None,
         parent_span_id: str | None = None,
         start_to_close_timeout: timedelta = timedelta(seconds=600),
@@ -227,6 +231,10 @@ async def run_agent_auto_send(
         Returns:
             Union[SerializableRunResult, RunResult]: SerializableRunResult when in Temporal, RunResult otherwise.
         """
+        # Default to empty list if not provided
+        if mcp_server_params is None:
+            mcp_server_params = []
+
         if in_temporal_workflow():
             params = RunAgentAutoSendParams(
                 trace_id=trace_id,
@@ -283,9 +291,9 @@ async def run_agent_auto_send(
     async def run_agent_streamed(
         self,
         input_list: list[dict[str, Any]],
-        mcp_server_params: list[StdioServerParameters],
         agent_name: str,
         agent_instructions: str,
+        mcp_server_params: list[StdioServerParameters] | None = None,
         trace_id: str | None = None,
         parent_span_id: str | None = None,
         handoff_description: str | None = None,
@@ -340,6 +348,10 @@ async def run_agent_streamed(
         Raises:
             ValueError: If called from within a Temporal workflow
         """
+        # Default to empty list if not provided
+        if mcp_server_params is None:
+            mcp_server_params = []
+
         # Temporal workflows should use the auto_send variant
         if in_temporal_workflow():
             raise ValueError(
@@ -373,9 +385,9 @@ async def run_agent_streamed_auto_send(
         self,
         task_id: str,
         input_list: list[dict[str, Any]],
-        mcp_server_params: list[StdioServerParameters],
         agent_name: str,
         agent_instructions: str,
+        mcp_server_params: list[StdioServerParameters] | None = None,
         trace_id: str | None = None,
         parent_span_id: str | None = None,
         start_to_close_timeout: timedelta = timedelta(seconds=600),
@@ -426,6 +438,10 @@ async def run_agent_streamed_auto_send(
         Returns:
             Union[SerializableRunResultStreaming, RunResultStreaming]: SerializableRunResultStreaming when in Temporal, RunResultStreaming otherwise.
         """
+        # Default to empty list if not provided
+        if mcp_server_params is None:
+            mcp_server_params = []
+
         if in_temporal_workflow():
             params = RunAgentStreamedAutoSendParams(
                 trace_id=trace_id,

src/agentex/lib/sdk/fastacp/base/base_acp_server.py

@@ -363,8 +363,12 @@ async def message_send_wrapper(params: SendMessageParams):
             else:
                 # The client wants streaming, but the function is not an async generator, so we turn it into one and yield each TaskMessageContent as a StreamTaskMessageFull which will be streamed to the client by the Agentex server.
                 task_message_content_response = await fn(params)
-                if isinstance(task_message_content_response, list):
-                    task_message_content_list = task_message_content_response
+                # Handle None returns gracefully - treat as empty list
+                if task_message_content_response is None:
+                    task_message_content_list = []
+                elif isinstance(task_message_content_response, list):
+                    # Filter out None values from lists
+                    task_message_content_list = [content for content in task_message_content_response if content is not None]
                 else:
                     task_message_content_list = [task_message_content_response]