Merge pull request #260 from veithly/fix/streaming-infrastructure

fix: streaming infrastructure — queue, loop condition, and timeout

Author: moonpyt

spoon_ai/agents/base.py

@@ -57,6 +57,10 @@ def __init__(self, maxsize: int = 0):
     async def put(self, item: Any) -> None:
         await self._queue.put(item)
 
+    def put_nowait(self, item: Any) -> None:
+        """Non-blocking put - delegates to the underlying asyncio.Queue."""
+        self._queue.put_nowait(item)
+
     async def get(self, timeout: float | None = 30.0) -> Any:
         """Get item with timeout and fair access"""
         consumer_id = id(asyncio.current_task())
@@ -104,7 +108,7 @@ class BaseAgent(BaseModel, ABC):
     # Thread-safe replacements
     output_queue: ThreadSafeOutputQueue = Field(default_factory=ThreadSafeOutputQueue, description="Thread-safe output queue")
     task_done: asyncio.Event = Field(default_factory=asyncio.Event, description="The signal of agent run done")
-    
+
     # Callback system
     callbacks: list[BaseCallbackHandler] = Field(default_factory=list, description="Callback handlers for monitoring")
 
@@ -335,7 +339,7 @@ async def add_message_with_image(
 
         if not image_url and not image_data:
             raise ValueError("Either image_url or image_data must be provided")
-        
+
         # Validate image_data is not empty (if provided)
         # Three upload methods:
         # - Method 1: image_data (base64) - image_data must have a value
@@ -349,7 +353,7 @@ async def add_message_with_image(
             # Check if only whitespace (empty after strip)
             if not image_data.strip():
                 raise ValueError("image_data cannot be empty (only whitespace). If you want to use URL-based images, use image_url parameter instead.")
-        
+
         # Validate image_url format if provided
         # image_url supports both external URLs (way 2) and data URLs (way 3)
         if image_url:
@@ -366,7 +370,7 @@ async def add_message_with_image(
                         f"Invalid image URL format: {image_url}. "
                         f"Must be a valid HTTP/HTTPS URL (for external images) or data URL (for embedded images)."
                     )
-        
+
         # No MIME type validation - pass through all types to LLM providers
 
         content_blocks: list[ContentBlock] = [TextContent(text=text)]
@@ -1055,7 +1059,12 @@ async def stream(self, timeout: float | None = None):
         try:
             self._active_operations.add(stream_id)
 
-            while not (self.task_done.is_set() or self.output_queue.empty()):
+            # Continue streaming while the task is still running OR the queue
+            # has remaining items to drain.  The previous condition
+            # ``not (done or empty)`` was equivalent to ``not done and not empty``
+            # which would exit immediately when the queue started empty even
+            # though the background task had not yet produced output.
+            while not (self.task_done.is_set() and self.output_queue.empty()):
                 try:
                     # Create tasks for queue and done event
                     queue_task = asyncio.create_task(

spoon_ai/agents/spoon_react.py

@@ -187,7 +187,12 @@ async def _ensure_x402_tools(self) -> None:
 
         self._x402_tools_initialized = True
 
-    async def run(self, request: Optional[str] = None) -> str:
+    async def run(self, request: Optional[str] = None, timeout: Optional[float] = None) -> str:
         """Ensure prompts reflect current tools before running."""
         self._refresh_prompts()
-        return await super().run(request)
+        kwargs: dict = {}
+        if request is not None:
+            kwargs["request"] = request
+        if timeout is not None:
+            kwargs["timeout"] = timeout
+        return await super().run(**kwargs)

spoon_ai/agents/spoon_react_skill.py

@@ -89,7 +89,7 @@ def __init__(self, **kwargs):
 
         self._skill_manager_initialized = True
 
-    async def run(self, request: Optional[str] = None) -> str:
+    async def run(self, request: Optional[str] = None, timeout: Optional[float] = None) -> str:
         """
         Execute agent with per-turn auto skill activation.
 
@@ -102,15 +102,20 @@ async def run(self, request: Optional[str] = None) -> str:
 
         Args:
             request: User request/message
+            timeout: Optional timeout in seconds
 
         Returns:
             Agent response
         """
-
         async def _runner(req: Optional[str]) -> str:
             # SpoonReactAI.run() rebuilds prompts from available_tools.
             # Ensure skill tools are synced first, then delegate to parent.
-            return await super(SpoonReactSkill, self).run(req)
+            kwargs: dict = {}
+            if req is not None:
+                kwargs["request"] = req
+            if timeout is not None:
+                kwargs["timeout"] = timeout
+            return await super(SpoonReactSkill, self).run(**kwargs)
 
         return await self._run_with_auto_skills(request, _runner)
 

spoon_ai/agents/toolcall.py

@@ -215,6 +215,9 @@ def convert_mcp_tool(tool: MCPTool) -> dict:
             logger.info(f"🏁 {self.name} terminating due to finish_reason signals (no tool calls)")
             self.state = AgentState.FINISHED
             await self.add_message("assistant", response.content or "Task completed")
+            # Emit content to output queue for streaming consumers
+            if self.output_queue:
+                self.output_queue.put_nowait({"content": response.content or "Task completed"})
             # Set a flag to indicate finish_reason termination and store the content
             self._finish_reason_terminated = True
             self._final_response_content = response.content or "Task completed"

spoon_ai/skills/loader.py

@@ -109,22 +109,25 @@ def discover(self) -> List[Path]:
         Returns:
             List of paths to SKILL.md files
         """
-        skill_files = []
+        skill_files: List[Path] = []
+        seen_resolved: set = set()
 
         for base_path in self._paths:
             if not base_path.exists():
                 continue
 
-            # Find all SKILL.md files (case-insensitive on Windows)
-            for skill_md in base_path.rglob("SKILL.md"):
-                skill_files.append(skill_md)
-                logger.debug(f"Discovered skill: {skill_md}")
-
-            # Also check for skill.md (lowercase)
-            for skill_md in base_path.rglob("skill.md"):
-                if skill_md not in skill_files:
-                    skill_files.append(skill_md)
-                    logger.debug(f"Discovered skill: {skill_md}")
+            # Use a single glob and case-insensitive name check to avoid
+            # duplicates on Windows (where rglob("SKILL.md") and
+            # rglob("skill.md") may return the same files).
+            for md_file in base_path.rglob("*.md"):
+                if md_file.name.lower() != "skill.md":
+                    continue
+                resolved = md_file.resolve()
+                if resolved in seen_resolved:
+                    continue
+                seen_resolved.add(resolved)
+                skill_files.append(md_file)
+                logger.debug(f"Discovered skill: {md_file}")
 
         return skill_files
 
@@ -141,7 +144,10 @@ def parse(self, file_path: Path) -> Tuple[SkillMetadata, str]:
         Raises:
             ValueError: If file format is invalid
         """
-        content = file_path.read_text(encoding='utf-8')
+        # Use utf-8-sig to strip BOM if present (common on Windows editors)
+        content = file_path.read_text(encoding='utf-8-sig')
+        # Normalize line endings for regex reliability (\r\n → \n)
+        content = content.replace('\r\n', '\n').replace('\r', '\n')
 
         match = self.FRONTMATTER_PATTERN.match(content)
         if not match:

spoon_ai/skills/models.py

@@ -52,6 +52,12 @@ class SkillScript(BaseModel):
     timeout: int = Field(default=30, ge=1, le=600, description="Execution timeout in seconds")
     working_directory: Optional[str] = Field(default=None, description="Working directory override")
 
+    # Structured input schema (mirrors JSON Schema for script stdin)
+    input_schema: Optional[Dict[str, Any]] = Field(
+        default=None,
+        description="JSON Schema describing the structured input the script expects via stdin"
+    )
+
     # Lifecycle hooks
     run_on_activation: bool = Field(default=False, description="Run when skill activates")
     run_on_deactivation: bool = Field(default=False, description="Run when skill deactivates")

spoon_ai/skills/script_tool.py

@@ -5,6 +5,7 @@
 AI decides how to use scripts - users only control whether scripts are allowed.
 """
 
+import json
 import logging
 from typing import Any, Dict, List, Optional
 
@@ -22,7 +23,10 @@ class ScriptTool(BaseTool):
     Tool wrapper for skill scripts.
 
     Exposes a SkillScript as a callable tool that agents can invoke.
-    The AI decides what input to provide - there's no fixed parameter schema.
+    When the script defines an ``input_schema``, the tool parameters are
+    derived from that schema so the LLM receives a structured contract.
+    Otherwise a generic ``input`` string parameter is used for backward
+    compatibility.
     """
 
     name: str = Field(..., description="Tool name")
@@ -33,6 +37,7 @@ class ScriptTool(BaseTool):
     script: SkillScript = Field(..., exclude=True)
     skill_name: str = Field(..., exclude=True)
     working_directory: Optional[str] = Field(default=None, exclude=True)
+    _uses_structured_schema: bool = False
 
     def __init__(
         self,
@@ -55,17 +60,49 @@ def __init__(
         desc = script.description or f"Execute the '{script.name}' script"
         description = f"{desc} (Type: {script.type.value})"
 
-        # Simple parameter schema - just optional input
-        parameters = {
-            "type": "object",
-            "properties": {
-                "input": {
-                    "type": "string",
-                    "description": "Optional input text to pass to the script via stdin"
+        # Derive parameter schema from script.input_schema when available (#8)
+        uses_structured = False
+        if script.input_schema and isinstance(script.input_schema, dict):
+            schema_type = script.input_schema.get("type", "object")
+            # Tool/function calling interfaces expect top-level object schema.
+            # If skill metadata declares non-object type, degrade gracefully.
+            if schema_type != "object":
+                logger.warning(
+                    "Script '%s' in skill '%s' has non-object input_schema.type=%s; "
+                    "falling back to generic object schema",
+                    script.name,
+                    skill_name,
+                    schema_type,
+                )
+                parameters = {
+                    "type": "object",
+                    "properties": {
+                        "input": {
+                            "type": "string",
+                            "description": "Optional input text to pass to the script via stdin"
+                        }
+                    },
+                    "required": []
                 }
-            },
-            "required": []
-        }
+            else:
+                parameters = {
+                    "type": "object",
+                    "properties": script.input_schema.get("properties", {}),
+                    "required": script.input_schema.get("required", []),
+                }
+                uses_structured = True
+        else:
+            # Fallback: generic optional input string (backward compat)
+            parameters = {
+                "type": "object",
+                "properties": {
+                    "input": {
+                        "type": "string",
+                        "description": "Optional input text to pass to the script via stdin"
+                    }
+                },
+                "required": []
+            }
 
         super().__init__(
             name=tool_name,
@@ -75,14 +112,20 @@ def __init__(
             skill_name=skill_name,
             working_directory=working_directory
         )
+        object.__setattr__(self, "_uses_structured_schema", uses_structured)
 
     async def execute(self, input: Optional[str] = None, **kwargs) -> str:
         """
         Execute the script.
 
+        When the script declares an ``input_schema``, the LLM's structured
+        kwargs are serialized to JSON and piped to stdin.  For legacy scripts
+        that only declare a generic ``input`` string, the raw value is passed
+        through as-is.
+
         Args:
-            input: Optional input text to pass to script via stdin
-            **kwargs: Additional arguments (ignored)
+            input: Optional input text (legacy path)
+            **kwargs: Structured arguments matching input_schema
 
         Returns:
             Script output as string
@@ -91,9 +134,24 @@ async def execute(self, input: Optional[str] = None, **kwargs) -> str:
 
         logger.debug(f"ScriptTool '{self.name}' executing")
 
+        # Decide what to send to the script on stdin
+        if self._uses_structured_schema:
+            # Build a JSON payload from all kwargs (including 'input' if present)
+            payload: Dict[str, Any] = {}
+            if input is not None:
+                payload["input"] = input
+            payload.update(kwargs)
+            input_text = json.dumps(payload, ensure_ascii=False)
+        else:
+            # Legacy path: plain string or try JSON passthrough
+            input_text = input
+            if input_text is None and kwargs:
+                # Model may have sent structured args despite generic schema
+                input_text = json.dumps(kwargs, ensure_ascii=False)
+
         result: ScriptResult = await executor.execute(
             script=self.script,
-            input_text=input,
+            input_text=input_text,
             working_directory=self.working_directory
         )
 

spoon_ai/tools/mcp_tool.py

@@ -142,11 +142,20 @@ async def _fetch_and_set_parameters(self):
                     async with self.get_session() as session:
                         tools = await asyncio.wait_for(session.list_tools(), timeout=self._connection_timeout)
                         if tools:
+                            # Try exact name match first; fall back to first tool
                             target_tool = None
                             for tool in tools:
                                 if getattr(tool, 'name', '') == self.name:
                                     target_tool = tool
                                     break
+                            # Also try matching against _server_name (the original
+                            # alias) so that the first call still resolves after
+                            # a previous expand_server_tools renamed `self.name`.
+                            if not target_tool and hasattr(self, '_server_name'):
+                                for tool in tools:
+                                    if getattr(tool, 'name', '') == self._server_name:
+                                        target_tool = tool
+                                        break
 
                             if not target_tool and tools:
                                 target_tool = tools[0]
@@ -350,6 +359,50 @@ async def call_mcp_tool(self, tool_name: str, **kwargs):
             logger.error(f"MCP tool '{tool_name}' call failed: {e}")
             raise RuntimeError(f"MCP tool '{tool_name}' execution failed: {str(e)}") from e
 
+    async def expand_server_tools(self) -> List["MCPTool"]:
+        """Expand this single MCPTool (one-per-server) into one MCPTool per
+        real server tool.  Each returned tool shares the same MCP transport
+        config and delegates execution to ``call_mcp_tool(real_name)``.
+
+        If the server is unreachable or returns no tools, an empty list is
+        returned (callers should keep the original proxy as fallback).
+
+        Returns:
+            List of MCPTool instances, one per discovered server tool.
+        """
+        server_tools = await self.list_available_tools()
+        if not server_tools:
+            return []
+
+        expanded: List[MCPTool] = []
+        server_name = self.name  # Original alias (e.g. "filesystem")
+
+        for tool_info in server_tools:
+            real_name = tool_info.get("name", "")
+            if not real_name:
+                continue
+            desc = tool_info.get("description", f"MCP tool from {server_name}")
+            schema = tool_info.get("inputSchema") or {
+                "type": "object", "properties": {}, "required": []
+            }
+
+            child = MCPTool(
+                name=real_name,
+                description=desc,
+                parameters=schema,
+                mcp_config=self.mcp_config,
+            )
+            child._parameters_loaded = True  # Already resolved
+            # Store original server alias so execute fallback can find it
+            object.__setattr__(child, '_server_name', server_name)
+            expanded.append(child)
+
+        logger.info(
+            f"Expanded MCP server '{server_name}' into {len(expanded)} tools: "
+            f"{[t.name for t in expanded]}"
+        )
+        return expanded
+
     async def list_available_tools(self) -> list:
         """List available tools from the MCP server."""
         try: