Added Shell

agent/__init__.py

@@ -1,6 +1,7 @@
 from .agent import Agent
-from .tools import Tool, tool, fetch_webpage, FETCH_WEBPAGE_TOOL
+from .tools import Tool, tool, fetch_webpage, FETCH_WEBPAGE_TOOL, SHELL_TOOL
 from .mcp import MCPClient, mcp_tool, load_mcp_tools, load_all_mcp_tools, MCP_SERVERS
+from .shell import ShellManager, get_shell_manager
 
 __all__ = [
     "Agent",
@@ -8,9 +9,12 @@ __all__ = [
     "tool",
     "fetch_webpage",
     "FETCH_WEBPAGE_TOOL",
+    "SHELL_TOOL",
     "MCPClient",
     "mcp_tool",
     "load_mcp_tools",
     "load_all_mcp_tools",
     "MCP_SERVERS",
+    "ShellManager",
+    "get_shell_manager",
 ]

agent/agent.py

@@ -14,7 +14,7 @@ from .mcp import MCPClient, load_mcp_tools, load_all_mcp_tools
 #   {"type": "text",     "content": "partial text"}
 #   {"type": "reasoning","content": "model thinking"}
 #   {"type": "tool_call", "name": str, "arguments": '{"url":"..."}'}
-#   {"type": "tool_output","name": str, "content": "result"}
+#   {"type": "tool_output","name": str, "arguments": str, "content": "result", "partial": bool}
 #   {"type": "done",     "content": "full assistant response"}
 #   {"type": "error",    "content": "error message"}
 # ---------------------------------------------------------------------------
@@ -42,6 +42,7 @@ class Agent:
         self._final_tool_name: str | None = None
         self._max_iterations = max_iterations
         self.system_prompt = system_prompt
+        self._tool_results: dict[str, str] = {}  # tool_call_id → full result
 
     # ------------------------------------------------------------------
     # Tool registration
@@ -50,6 +51,58 @@ class Agent:
     def register_tool(self, tool: Tool) -> None:
         self._tools.append(tool)
 
+    def register_read_tool(self) -> None:
+        """Register a tool to read truncated tool responses by line range."""
+        results_cache = self._tool_results
+
+        def _read(tool_call_id: str, start_line: int, num_lines: int = 50) -> str:
+            """Read more lines from a truncated tool response.
+
+            tool_call_id (required): The tool_call_id from the truncated response
+            start_line (required): Line number to start reading from
+            num_lines: Number of lines to read (default 50)
+            """
+            full = results_cache.get(tool_call_id)
+            if full is None:
+                return f"Error: No result found for tool_call_id '{tool_call_id}'"
+            lines = full.split("\n")
+            total = len(lines)
+            if start_line >= total:
+                return f"Error: start_line {start_line} >= total lines {total}"
+            end = min(start_line + num_lines, total)
+            chunk = "\n".join(lines[start_line:end])
+            remaining = total - end
+            header = f"Lines {start_line}-{end} of {total}"
+            if remaining > 0:
+                header += f" ({remaining} lines remaining)"
+            return f"{header}\n\n{chunk}"
+
+        self._tools.append(
+            Tool(
+                name="read_tool_response",
+                description="Read more lines from a truncated tool response. Use when a previous tool output was truncated.",
+                parameters={
+                    "type": "object",
+                    "properties": {
+                        "tool_call_id": {
+                            "type": "string",
+                            "description": "The tool_call_id from the truncated response",
+                        },
+                        "start_line": {
+                            "type": "integer",
+                            "description": "Line number to start reading from (0-indexed)",
+                        },
+                        "num_lines": {
+                            "type": "integer",
+                            "description": "Number of lines to read (default 50)",
+                        },
+                    },
+                    "required": ["tool_call_id", "start_line"],
+                },
+                handler=_read,
+            )
+        )
+
     def register_mcp(self, url: str, headers: dict[str, str] | None = None) -> list[Tool]:
         """Connect to an MCP server and register all its tools.
 
@@ -239,30 +292,108 @@ class Agent:
                     tool_obj = next(
                         (t for t in self._tools if t.name == tname), None
                     )
-                    if tool_obj:
+
+                    if tool_obj is None:
+                        result_str = f"Error: Tool '{tname}' not found"
+                        self._tool_results[tc_spec["id"]] = result_str
+                        yield {
+                            "type": "tool_output",
+                            "name": tname,
+                            "arguments": tc_spec["function"]["arguments"],
+                            "content": result_str,
+                        }
+                        messages.append({
+                            "role": "tool",
+                            "tool_call_id": tc_spec["id"],
+                            "content": result_str,
+                        })
+                        continue
+
+                    # Streamable tool — yield partial results
+                    if tool_obj.streamable:
+                        accumulated = ""
+                        last_chunk = ""
                         try:
-                            result = tool_obj.run(**targs)
+                            for chunk in tool_obj.stream(**targs):
+                                accumulated += chunk
+                                last_chunk = chunk
+                                # Truncate display but keep full for read_tool_response
+                                display = accumulated
+                                if len(display) > 5_000:
+                                    lines = display.split("\n")
+                                    char_count = 0
+                                    cut_line = 0
+                                    for i, line in enumerate(lines):
+                                        char_count += len(line) + 1
+                                        if char_count > 5_000:
+                                            cut_line = i
+                                            break
+                                    display = "\n".join(lines[:cut_line])
+                                yield {
+                                    "type": "tool_output",
+                                    "name": tname,
+                                    "arguments": tc_spec["function"]["arguments"],
+                                    "content": display,
+                                    "partial": True,
+                                }
+                            # Mark final yield as non-partial
+                            display = accumulated
+                            if len(display) > 5_000:
+                                lines = display.split("\n")
+                                char_count = 0
+                                cut_line = 0
+                                for i, line in enumerate(lines):
+                                    char_count += len(line) + 1
+                                    if char_count > 5_000:
+                                        cut_line = i
+                                        break
+                                display = "\n".join(lines[:cut_line])
+                            yield {
+                                "type": "tool_output",
+                                "name": tname,
+                                "arguments": tc_spec["function"]["arguments"],
+                                "content": display,
+                                "partial": False,
+                            }
+                            result_str = accumulated
                         except Exception as e:
-                            result = f"Error executing {tname}: {e}"
+                            result_str = f"Error executing {tname}: {e}"
                     else:
-                        result = f"Error: Tool '{tname}' not found"
+                        # Regular tool — single result
+                        try:
+                            result_str = str(tool_obj.run(**targs))
+                        except Exception as e:
+                            result_str = f"Error executing {tname}: {e}"
 
-                    result_str = str(result)
+                    # Store full result and truncate for message history
+                    self._tool_results[tc_spec["id"]] = result_str
+                    lines = result_str.split("\n")
                     if len(result_str) > 5_000:
-                        result_str = result_str[:5_000] + "\n...[truncated]"
-                    yield {
-                        "type": "tool_output",
-                        "name": tname,
-                        "content": result_str,
-                    }
-
-                    messages.append(
-                        {
-                            "role": "tool",
-                            "tool_call_id": tc_spec["id"],
+                        char_count = 0
+                        cut_line = 0
+                        for i, line in enumerate(lines):
+                            char_count += len(line) + 1
+                            if char_count > 5_000:
+                                cut_line = i
+                                break
+                        truncated = "\n".join(lines[:cut_line])
+                        remaining = len(lines) - cut_line
+                        result_str = f"{truncated}\n\n...[{remaining} lines truncated — use read_tool_response tool_call_id=\"{tc_spec['id']}\" start_line={cut_line} to read more]"
+
+                    # Final tool_output (non-partial) for history
+                    if not tool_obj.streamable:
+                        yield {
+                            "type": "tool_output",
+                            "name": tname,
+                            "arguments": tc_spec["function"]["arguments"],
                             "content": result_str,
                         }
-                    )
+
+                    messages.append({
+                        "role": "tool",
+                        "tool_call_id": tc_spec["id"],
+                        "content": result_str,
+                    })
 
                 continue  # Loop back — model can call more tools or final_message
 

agent/shell.py

@@ -0,0 +1,180 @@
+"""Shell manager for running interactive subprocesses with streaming output."""
+
+from __future__ import annotations
+
+import subprocess
+import threading
+import time
+from dataclasses import dataclass, field
+from typing import IO
+
+
+@dataclass
+class ShellSession:
+    """A running shell session."""
+
+    session_id: str
+    process: subprocess.Popen[str]
+    output_buffer: list[str] = field(default_factory=list)
+    last_read_pos: int = 0
+    started_at: float = field(default_factory=time.time)
+    closed: bool = False
+
+    @property
+    def pid(self) -> int | None:
+        return self.process.pid
+
+    @property
+    def returncode(self) -> int | None:
+        return self.process.returncode
+
+    def read_new_output(self) -> str:
+        """Read new output since last read."""
+        new_lines = self.output_buffer[self.last_read_pos :]
+        self.last_read_pos = len(self.output_buffer)
+        return "".join(new_lines)
+
+    def get_full_output(self) -> str:
+        """Get all output."""
+        return "".join(self.output_buffer)
+
+    def is_running(self) -> bool:
+        """Check if process is still running."""
+        return self.process.poll() is None
+
+    def close(self) -> None:
+        """Close the session."""
+        if not self.closed:
+            self.closed = True
+            try:
+                self.process.terminate()
+                self.process.wait(timeout=5)
+            except Exception:
+                try:
+                    self.process.kill()
+                except Exception:
+                    pass
+
+
+class ShellManager:
+    """Manages multiple shell sessions with streaming output."""
+
+    def __init__(self, default_timeout: float = 15.0) -> None:
+        self.sessions: dict[str, ShellSession] = {}
+        self._lock = threading.Lock()
+        self.default_timeout = default_timeout
+
+    def start(
+        self,
+        session_id: str,
+        command: str,
+        cwd: str | None = None,
+        env: dict[str, str] | None = None,
+    ) -> ShellSession:
+        """Start a new shell session."""
+        with self._lock:
+            # Close existing session with same ID
+            if session_id in self.sessions:
+                self.sessions[session_id].close()
+
+            # Use shell=True for interactive commands
+            process = subprocess.Popen(
+                command,
+                shell=True,
+                stdin=subprocess.PIPE,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.STDOUT,
+                text=True,
+                bufsize=1,
+                cwd=cwd,
+                env=env,
+            )
+
+            session = ShellSession(session_id=session_id, process=process)
+            self.sessions[session_id] = session
+
+            # Start output reader thread
+            thread = threading.Thread(
+                target=self._read_output, args=(session,), daemon=True
+            )
+            thread.start()
+
+            return session
+
+    def _read_output(self, session: ShellSession) -> None:
+        """Read output from process in background thread."""
+        assert session.process.stdout is not None
+        try:
+            for line in session.process.stdout:
+                with self._lock:
+                    session.output_buffer.append(line)
+        except Exception:
+            pass
+        finally:
+            with self._lock:
+                session.closed = True
+
+    def send_input(self, session_id: str, input_text: str) -> bool:
+        """Send input to a running session."""
+        with self._lock:
+            session = self.sessions.get(session_id)
+            if session is None or not session.is_running():
+                return False
+            if session.process.stdin is None:
+                return False
+            try:
+                session.process.stdin.write(input_text + "\n")
+                session.process.stdin.flush()
+                return True
+            except Exception:
+                return False
+
+    def get_output(self, session_id: str) -> str | None:
+        """Get full output from a session."""
+        with self._lock:
+            session = self.sessions.get(session_id)
+            if session is None:
+                return None
+            return session.get_full_output()
+
+    def poll_output(self, session_id: str) -> str | None:
+        """Get new output since last poll."""
+        with self._lock:
+            session = self.sessions.get(session_id)
+            if session is None:
+                return None
+            return session.read_new_output()
+
+    def is_running(self, session_id: str) -> bool:
+        """Check if a session is still running."""
+        with self._lock:
+            session = self.sessions.get(session_id)
+            if session is None:
+                return False
+            return session.is_running()
+
+    def close(self, session_id: str) -> None:
+        """Close a session."""
+        with self._lock:
+            session = self.sessions.get(session_id)
+            if session:
+                session.close()
+
+    def close_all(self) -> None:
+        """Close all sessions."""
+        with self._lock:
+            for session in self.sessions.values():
+                session.close()
+            self.sessions.clear()
+
+
+# Global shell manager instance
+_shell_manager: ShellManager | None = None
+
+
+def get_shell_manager() -> ShellManager:
+    """Get or create the global shell manager."""
+    global _shell_manager
+    if _shell_manager is None:
+        _shell_manager = ShellManager()
+    return _shell_manager

agent/tools.py

@@ -1,5 +1,5 @@
 from markitdown import MarkItDown
-from typing import Any, Callable, get_type_hints
+from typing import Any, Callable, Generator, get_type_hints
 import inspect
 import requests
 
@@ -20,12 +20,18 @@ class Tool:
     """A callable tool the agent can invoke."""
 
     def __init__(
-        self, name: str, description: str, parameters: dict, handler: Callable[..., str]
+        self,
+        name: str,
+        description: str,
+        parameters: dict,
+        handler: Callable[..., str],
+        streamable: bool = False,
     ) -> None:
         self.name = name
         self.description = description
         self.parameters = parameters
         self.handler = handler
+        self.streamable = streamable
 
     def to_openai_spec(self) -> dict:
         return {
@@ -40,6 +46,20 @@ class Tool:
     def run(self, **kwargs: Any) -> str:
         return self.handler(**kwargs)
 
+    def stream(self, **kwargs: Any) -> Generator[str, None, None]:
+        """Yield partial results for streamable tools.
+
+        Override in subclasses or use streamable=True with a generator handler.
+        """
+        if self.streamable and callable(self.handler):
+            result = self.handler(**kwargs)
+            if isinstance(result, Generator):
+                yield from result
+            else:
+                yield str(result)
+        else:
+            yield self.handler(**kwargs)
+
 
 def _parse_docstring(docstring: str) -> tuple[str, dict[str, tuple[bool, str]]]:
     """Parse a tool docstring into description and param metadata.
@@ -147,3 +167,111 @@ def fetch_webpage(url: str) -> str:
         return md.convert(url).text_content
 
 FETCH_WEBPAGE_TOOL = fetch_webpage  # @tool already makes it a Tool instance
+
+
+# ---------------------------------------------------------------------------
+# Shell tool (streamable)
+# ---------------------------------------------------------------------------
+
+import time
+import uuid as _uuid
+from .shell import get_shell_manager
+
+
+def _shell_handler(
+    command: str,
+    session_id: str = "",
+    input_text: str = "",
+    timeout: float = 15.0,
+) -> Generator[str, None, None]:
+    """Run a shell command with streaming output.
+
+    command (required): The shell command to execute
+    session_id: Session ID to send input to (omit to auto-generate)
+    input_text: Text to send to running session's stdin
+    timeout: Seconds between output updates (default 15)
+    """
+    manager = get_shell_manager()
+
+    # Check if session_id refers to an existing session
+    existing_session = session_id and session_id in manager.sessions
+
+    # If session exists and has input, send it
+    if existing_session and input_text:
+        sent = manager.send_input(session_id, input_text)
+        if not sent:
+            yield f"Error: Session '{session_id}' closed or cannot accept input"
+            return
+        time.sleep(0.5)
+        output = manager.poll_output(session_id)
+        if output:
+            yield f"Sent input. New output:\n{output}"
+        else:
+            yield "Input sent (no new output yet)"
+        return
+
+    # If session exists without input, return current output
+    if existing_session:
+        output = manager.get_output(session_id)
+        if output is None:
+            yield f"Error: Session '{session_id}' not found"
+            return
+        running = manager.is_running(session_id)
+        status = "running" if running else f"exited (code {manager.sessions[session_id].returncode})"
+        yield f"Session {session_id} [{status}]:\n{output}"
+        return
+
+    # Start new command (with provided session_id or auto-generated)
+    sid = session_id or str(_uuid.uuid4())[:8]
+    session = manager.start(sid, command)
+    yield f"Started session {sid} (PID {session.pid})"
+
+    # Stream output — poll frequently, yield when there's new output
+    last_yield = time.time()
+    while session.is_running():
+        time.sleep(0.5)
+        output = session.read_new_output()
+        if output:
+            print(f"Debug: New output for session {sid}:\n{output}")
+            yield f"[{sid}] Output:\n{output}"
+            last_yield = time.time()
+
+    # Final output
+    time.sleep(0.2)
+    final = session.read_new_output()
+    code = session.process.returncode
+    status = f"exited with code {code}" if code is not None else "exited"
+    if final:
+        yield f"[{sid}] Final ({status}):\n{final}"
+    else:
+        yield f"[{sid}] {status}"
+
+
+SHELL_TOOL = Tool(
+    name="shell",
+    description="Run shell commands with streaming output. Supports interactive sessions — send input to running commands.",
+    parameters={
+        "type": "object",
+        "properties": {
+            "command": {
+                "type": "string",
+                "description": "The shell command to execute",
+            },
+            "session_id": {
+                "type": "string",
+                "description": "Session ID to send input to or get output from (omit to start new command)",
+            },
+            "input_text": {
+                "type": "string",
+                "description": "Text to send to running session's stdin",
+            },
+            "timeout": {
+                "type": "number",
+                "description": "Seconds between output updates (default 15)",
+            },
+        },
+        "required": ["command"],
+    },
+    handler=_shell_handler,
+    streamable=True,
+)

app.py

@@ -2,7 +2,7 @@ import time
 import uuid
 import gradio as gr
 from dotenv import load_dotenv
-from agent import Agent, FETCH_WEBPAGE_TOOL
+from agent import Agent, FETCH_WEBPAGE_TOOL, SHELL_TOOL
 import os
 from pathlib import Path
 
@@ -31,7 +31,9 @@ agent = Agent(
     system_prompt=_SYSTEM_PROMPT,
 )
 agent.register_tool(FETCH_WEBPAGE_TOOL)
+agent.register_tool(SHELL_TOOL)
 agent.register_all_mcp()
+agent.register_read_tool()
 agent.register_final_message_tool()
 
 # Load JS from external files
@@ -156,13 +158,15 @@ class GradioEvents:
                         # Grab the tool name from the existing message
                         tool_name = display_messages[tool_call_idx]["metadata"]["title"].split("Used tool ")[-1]
                         display_messages[tool_call_idx]["content"] = (
-                            f"```\n{tool_name}(...)\n```\n\n"
+                            f"```\n{tool_name}({ev['arguments']})\n```\n\n"
                             f"**Output:**\n```\n{cc}\n```"
                         )
                         display_messages[tool_call_idx]["metadata"] = {
                             "title": f"🛠️ {tool_name} — {len(ev['content'])} chars",
                         }
-                        tool_call_idx = None
+                        # Only clear tool_call_idx if NOT partial (final output)
+                        if not ev.get("partial"):
+                            tool_call_idx = None
 
                 elif t == "error":
                     display_messages.append({
@@ -343,7 +347,8 @@ with gr.Blocks(fill_width=True, title="Demo Chat") as demo:
                 elem_id="chatbot",
                 show_label=False,
                 buttons=[],
-                layout="bubble"
+                layout="bubble",
+                autoscroll=False
             )
             with gr.Row(elem_id="input-row"):
                 msg = gr.Textbox(