coding-agent¶
第5章 · Coding Agent 与代码生成 · 配套项目
chapter5/coding-agent
项目说明¶
Comprehensive Coding Agent - Pure Python Implementation¶
A production-ready AI coding agent built with Claude, implementing all techniques from Chapter 2 with pure Python tools - no command-line dependencies required!
🌟 Key Features¶
✅ Pure Python Implementation¶
All tools implemented without command-line dependencies:
- ❌ No grep, rg (ripgrep), find commands needed
- ❌ No dependency on system utilities
- ✅ 100% pure Python implementations
- ✅ Works on any system with Python 3.8+
- ✅ Especially designed for Mac users without command-line tools
🛠️ Complete Tool Suite¶
All 16 tools from tools.json fully implemented:
File Operations (Pure Python):
- Read - File reading with image/PDF/notebook support
- Write - File writing with auto lint checking
- Edit - Search and replace editing
- MultiEdit - Multiple edits in one operation
Search Tools (Pure Python, no rg/grep dependency):
- Grep - Pure Python regex search with full ripgrep feature parity
- Full regex support
- Case insensitive search
- Context lines (before/after/around)
- Line numbers
- Multiline mode
- Glob filtering
- File type filtering
- Multiple output modes
- Glob - File pattern matching
- LS - Directory listing
Shell Operations:
- Bash - Persistent shell sessions
- BashOutput - Background job output
- KillBash - Terminate shells
Project Management:
- TodoWrite - Task list management
- ExitPlanMode - Plan mode exit
Advanced:
- NotebookEdit - Jupyter notebook editing
- WebFetch - Web content fetching (stub)
- WebSearch - Web search (stub)
- Task - Sub-agent launcher (stub)
🧠 System Hint Techniques (Chapter 2)¶
- Timestamps: Every message and tool result timestamped
- Tool Call Counting: Warns after 3+ repeated calls
- TODO List Management: Explicit task tracking
- Detailed Error Information: Rich error context
- System State Awareness: Working directory, OS, Python version
- Environment Information: Dynamic state in context
🔧 Terminal Environment¶
- Persistent Shell Sessions: Commands in same shell
- Working Directory Tracking: Directory changes persist
- Background Execution: Long-running command support
✅ Auto Lint Detection¶
After Write/Edit/MultiEdit:
- Python syntax checking
- JavaScript/TypeScript checking
- Errors appear immediately in tool results
📁 Project Structure¶
coding-agent/
├── agent.py # Main agent implementation
├── system_state.py # System state tracking
├── tool_registry.py # Tool name → implementation mapping
├── tools/ # All tool implementations
│ ├── __init__.py
│ ├── base.py # Base tool class
│ ├── bash_tool.py # Shell execution
│ ├── bash_output_tool.py # Background job output
│ ├── kill_bash_tool.py # Shell termination
│ ├── read_tool.py # File reading
│ ├── write_tool.py # File writing
│ ├── edit_tool.py # File editing
│ ├── multi_edit_tool.py # Multiple edits
│ ├── grep_tool.py # 🔥 Pure Python regex search (no rg!)
│ ├── glob_tool.py # File pattern matching
│ ├── ls_tool.py # Directory listing
│ ├── todo_write_tool.py # TODO management
│ ├── exit_plan_mode_tool.py
│ ├── notebook_edit_tool.py
│ ├── web_fetch_tool.py
│ ├── web_search_tool.py
│ ├── task_tool.py
│ └── shell_session.py # Shell session management
├── tools.json # Tool definitions
├── system-prompt.md # System prompt
├── config.py # Configuration
├── requirements.txt # Dependencies
└── README.md # This file
🚀 Installation¶
# Navigate to project directory
cd /Users/boj/ai-agent-book/projects/week5/coding-agent
# Install dependencies
pip install -r requirements.txt
# Set up environment
cp .env.example .env
# Edit .env and configure your provider
Configuration¶
Edit .env file:
# Choose your provider (anthropic, openai, or openrouter)
PROVIDER=anthropic
# Add API key for your chosen provider
ANTHROPIC_API_KEY=sk-ant-api03-...
# or
OPENROUTER_API_KEY=sk-or-v1-...
# or
OPENAI_API_KEY=sk-...
# Select model appropriate for your provider
DEFAULT_MODEL=claude-sonnet-5
See PROVIDERS.md for detailed provider configuration guide.
Requirements¶
Core dependencies:
- Python 3.8+
- anthropic - For Anthropic API
- openai - For OpenAI/OpenRouter API
- python-dotenv - For configuration
Optional (for enhanced features):
- PyPDF2 - For PDF reading
- requests, beautifulsoup4, html2text - For WebFetch
No command-line tools needed! Works on macOS without Homebrew packages.
Supported Providers¶
- Anthropic - Direct Claude API access
- OpenRouter - Access to Claude, GPT, Gemini, Llama, and more
- OpenAI - Direct GPT API access
The agent automatically handles the different API formats for each provider.
OpenRouter as a universal fallback¶
You do not need a direct Anthropic or OpenAI key to run the agent. If the
requested direct provider's key is missing, the agent transparently falls back
to OpenRouter (via the OpenAI-compatible SDK) as long as
OPENROUTER_API_KEY is set:
PROVIDER=anthropicwithANTHROPIC_API_KEY→ Anthropic SDK, unchanged (default behavior).PROVIDER=anthropicwithoutANTHROPIC_API_KEY(butOPENROUTER_API_KEYset) → routed through OpenRouter.PROVIDER=openaiwithOPENAI_API_KEY→ OpenAI SDK, unchanged.PROVIDER=openaiwithoutOPENAI_API_KEY(butOPENROUTER_API_KEYset) → routed through OpenRouter.
When falling back, the native model id is prefixed/mapped to an OpenRouter id:
| Requested model | OpenRouter id used |
|---|---|
claude-sonnet-* (e.g. claude-sonnet-5) |
anthropic/claude-sonnet-4.6 |
claude-haiku-* |
anthropic/claude-haiku-4.5 |
claude-opus-* / other claude-* |
anthropic/claude-opus-4.8 |
gpt-* / o1-* (e.g. gpt-5.6-luna) |
openai/<model> |
already prefixed (vendor/model) |
passed through unchanged |
So a user with only an OPENROUTER_API_KEY can run, e.g.:
# No ANTHROPIC_API_KEY needed — falls back to OpenRouter automatically
python main.py --provider anthropic --model claude-sonnet-5 -p "..."
# gpt-5.6-luna routed through OpenRouter (no OPENAI_API_KEY needed)
python main.py --provider openai --model gpt-5.6-luna -p "..."
Set PROVIDER=openrouter explicitly (with a vendor/model id) if you want to
target a specific OpenRouter model without any mapping.
📖 Usage¶
命令行入口（main.py）¶
main.py 是唯一推荐的入口，提供统一的 argparse 命令行界面。运行
python main.py --help 查看完整的中文帮助：
主要参数：
| 参数 | 说明 |
|---|---|
| （无参数） | 进入交互式对话（默认行为） |
-p, --prompt "任务" |
非交互模式：执行单个任务后退出，适合脚本 / CI |
--list-tools |
离线列出全部已注册工具及简介（无需 API Key，可用于自检） |
--provider {anthropic,openai,openrouter} |
临时覆盖 .env 中的 PROVIDER |
--model 模型名 |
临时覆盖 .env 中的 DEFAULT_MODEL |
--base-url URL |
临时覆盖 API Base URL（自建网关 / 兼容 OpenAI 的服务） |
--max-iterations N |
单个任务的最大 Agent 迭代轮数（默认 50） |
--no-color |
禁用彩色输出（无 TTY 时自动禁用） |
快速自检（离线，无需 API Key）¶
先确认工具集加载正常：
$ python main.py --list-tools
共 16 个工具：
Task Launch a new agent to handle complex, multi-step tasks autonomously.
Bash Executes a given bash command in a persistent shell session ...
Glob - Fast file pattern matching tool that works with any codebase size
Grep A powerful search tool built on ripgrep
...
端到端示例：让 Agent 完成一个真实编码任务¶
配置好 .env（见上文 Configuration）后，用一条命令让 Agent 创建并运行一个脚本：
成功时的终端输出结构大致如下（示意，实际轮次/调用次数取决于模型）：
✓ Agent initialized successfully
You: 创建 hello_world.py ...
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
🔧 Calling tool: Write
✓ Completed (call #1)
✓ No lint errors
File: hello_world.py
🔧 Calling tool: Bash
✓ Completed (call #2)
Output:
Hello, World!
Hello, Alice!
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
✅ Task completed!
Iterations: 2
Tool calls: 2
判定成功的标志：Agent 依次调用
Write写文件、Bash运行脚本， 终端出现脚本的真实输出，并以✅ Task completed!收尾。 （quickstart.py是同一任务的脚本化版本，可作对照。）
交互式对话（默认）¶
不带 -p 直接运行即进入交互式会话：
Features: - 🎨 Color-coded output for better readability - ⚡ Real-time streaming responses - 🔧 Live tool execution display - 📊 Built-in status command - 💬 Conversation history - 🔄 Reset command to start fresh
会话内命令（在对话中输入）：
- /help - Show help message
- /quit or /exit - Exit the CLI
- /reset - Reset conversation history
- /clear - Clear the screen
- /status - Show agent status (tool calls, TODOs, etc.)
其他示例脚本（均需 API Key）¶
python quickstart.py # 基础快速上手（与上文端到端示例同款任务）
python example_complex_task.py # 复杂多步任务
python example_with_system_hints.py # 系统提示（System Hint）技术演示
Programmatic Usage¶
from agent import CodingAgent
agent = CodingAgent(api_key="your-key")
for event in agent.run("List all Python files"):
if event["type"] == "text_delta":
print(event["delta"], end="", flush=True)
elif event["type"] == "done":
print("\n✅ Done!")
🔍 Pure Python Grep Implementation¶
The Grep tool is fully implemented in pure Python without any dependency on grep, rg, or other command-line tools. It provides all the features of ripgrep:
# Example: Search for pattern in files
{
"name": "Grep",
"input": {
"pattern": "def.*test",
"path": "/path/to/search",
"output_mode": "content",
"-i": True, # Case insensitive
"-C": 3, # 3 lines context
"-n": True, # Show line numbers
"glob": "*.py", # Only Python files
"multiline": False # Single line matching
}
}
Features:
- ✅ Full regex support (Python re module)
- ✅ Case insensitive search (-i)
- ✅ Context lines (-A, -B, -C)
- ✅ Line numbers (-n)
- ✅ Multiline mode
- ✅ Glob filtering (glob parameter)
- ✅ File type filtering (type parameter)
- ✅ Output modes: content, files_with_matches, count
- ✅ Head limit
- ✅ Recursive directory search
- ✅ Binary file skip
- ✅ Hidden file/directory skip
🏗️ Architecture¶
Modular Tool System¶
Each tool is implemented as a separate class inheriting from BaseTool:
class MyTool(BaseTool):
@property
def name(self) -> str:
return "MyTool"
def _execute_impl(self, params: Dict[str, Any]) -> Dict[str, Any]:
# Tool implementation
return {"result": "success"}
Tool Registry¶
ToolRegistry maps tool names to implementations:
registry = ToolRegistry()
tool = registry.get_tool("Grep", system_state)
result = tool.execute(params)
System State¶
SystemState tracks:
- Current working directory
- Tool call counts
- TODO list
- Shell sessions
- Environment info
System Hints¶
System hints are injected before each LLM call:
<system_hint>
# System State
Current Time: 2025-10-12 15:30:45
Working Directory: /Users/boj/coding-agent
OS: Darwin
Python: Python 3.11.5
# Tool Call Statistics
- Grep: 2 calls
- Write: 1 calls
# Current TODO List
✅ [1] Search for files (completed)
🔄 [2] Implement feature (in_progress)
⬜ [3] Write tests (pending)
</system_hint>
🎯 Design Principles¶
1. Pure Python Implementation¶
Why: Maximum portability and compatibility - Works on any system with Python - No Homebrew, apt, or other package managers needed - Consistent behavior across platforms
2. Modular Tool Architecture¶
Why: Maintainability and extensibility - Each tool is self-contained - Easy to add new tools - Easy to test individually - Clear separation of concerns
3. No Command-Line Dependencies¶
Why: Reliability and control
- Grep: Pure Python regex search
- Glob: Python's pathlib.glob()
- LS: Python's os and pathlib
- No subprocess calls for core functionality
- Full control over behavior
4. System Hints for Self-Awareness¶
Why: Better agent behavior - Prevents infinite loops (tool call counting) - Maintains task focus (TODO tracking) - Provides environmental context - Enables self-monitoring
📊 Comparison with Chapter 2¶
| Technique | Status | Implementation |
|---|---|---|
| Standard OpenAI Tool Format | ✅ | Anthropic SDK |
| Streaming Tool Calls | ✅ | Real-time JSON delta parsing |
| Parallel Tool Calls | ✅ | Multiple tools per response |
| Pure Python Tools | ✅ | No command-line dependencies |
| Grep without rg | ✅ | Pure Python regex search |
| Timestamps | ✅ | All messages/tools |
| Tool Call Counting | ✅ | Warns at 3+ |
| TODO List | ✅ | TodoWrite tool |
| System State | ✅ | Working dir, OS, Python |
| Persistent Shell | ✅ | Shell sessions |
| Auto Lint Detection | ✅ | After Write/Edit/MultiEdit |
🔧 Configuration¶
.env file:
# Required
ANTHROPIC_API_KEY=your_key_here
# Optional
DEFAULT_MODEL=claude-sonnet-5
MAX_ITERATIONS=50
MAX_TOKENS=8192
📝 Adding New Tools¶
- Create tool file in
tools/:
# tools/my_tool.py
from .base import BaseTool
class MyTool(BaseTool):
@property
def name(self) -> str:
return "MyTool"
def _execute_impl(self, params):
# Implementation
return {"result": "success"}
- Register in
tools/__init__.py:
- Add to
tool_registry.py:
- Add definition to
tools.json
🐛 Troubleshooting¶
"No module named 'tools'"¶
Make sure you're running from the project directory:
Grep not finding files¶
Check: - Path is correct - Pattern is valid regex - Glob pattern matches files - Files contain searchable text (not binary)
Shell commands fail¶
Ensure:
- Bash is available at /bin/bash
- Working directory exists
- Commands are properly quoted
🧪 Testing¶
Comprehensive test suite with 130+ tests covering all tool features.
Run Tests¶
# Install test dependencies
pip install -r requirements.txt
# Run all tests
pytest
# Run with coverage
pytest --cov=tools --cov-report=html
# Run specific tool tests
pytest tests/test_grep_tool.py
pytest tests/test_bash_tool.py
# Verbose output
pytest -v
Test Coverage¶
- 130+ tests across 14 test files
- 2,200+ lines of test code
- All major features from tools.json tested
- Integration tests for tool chaining and system hints
See tests/README.md for detailed test documentation.
🎓 Learning Path¶
- Start with examples: Run
python main.py(interactive CLI) - Run quickstart:
python quickstart.py - Explore system hints:
python example_with_system_hints.py - Study Grep implementation: See
tools/grep_tool.py - Run tests:
pytest -vto see all features in action - Read Chapter 2: Understand the theory
- Add custom tools: Extend the system
📚 References¶
- Chapter 2: Context Engineering (AI Agent Book)
- Tools specification:
tools.json - System prompt:
system-prompt.md - Anthropic Claude API: https://docs.anthropic.com/
🎉 Key Advantages¶
- No Dependencies on External Tools
- Pure Python implementation
- Works without rg, grep, find, etc.
-
Perfect for Mac users without Homebrew
-
Modular Architecture
- Each tool is a separate file
- Easy to understand and modify
-
Clear separation of concerns
-
Production Ready
- Comprehensive error handling
- Auto lint detection
- System hints for reliability
-
Streaming support for UX
-
Educational Value
- Learn how tools work internally
- Understand pure Python file operations
- See regex search implementation
- Study agent architecture patterns
📄 License¶
MIT
🤝 Contributing¶
This is an educational implementation. Feel free to adapt and extend!
Built with pure Python for maximum portability and learning! 🐍✨
源代码¶
agent.py¶
"""
Comprehensive Coding Agent - Modular implementation with pure Python tools
All tools implemented without command-line dependencies
"""
import os
import json
from typing import List, Dict, Any, Optional, Iterator
from pathlib import Path
from datetime import datetime
import anthropic
import openai
from system_state import SystemState
from tool_registry import ToolRegistry
class CodingAgent:
"""Main coding agent with streaming support and modular tool system"""
def __init__(self, api_key: str, model: str = "claude-sonnet-5", base_url: Optional[str] = None, provider: str = "anthropic"):
"""
Initialize coding agent
Args:
api_key: API key for the provider
model: Model identifier
base_url: Optional base URL for OpenRouter or other providers
provider: Provider name (anthropic, openai, openrouter)
"""
self.api_key = api_key
self.model = model
self.base_url = base_url
self.provider = provider.lower()
self.system_state = SystemState()
self.tool_registry = ToolRegistry()
self.messages: List[Dict[str, Any]] = []
self.tools = self._load_tools()
self.system_prompt = self._load_system_prompt()
# Initialize client based on provider
if self.provider == "anthropic":
# Use Anthropic SDK
self.client = anthropic.Anthropic(api_key=api_key)
self.client_type = "anthropic"
elif self.provider in ["openai", "openrouter"]:
# Use OpenAI SDK for both OpenAI and OpenRouter
# OpenRouter uses OpenAI-compatible API format
if base_url:
self.client = openai.OpenAI(api_key=api_key, base_url=base_url)
else:
self.client = openai.OpenAI(api_key=api_key)
self.client_type = "openai"
else:
raise ValueError(f"Unsupported provider: {provider}")
def _load_tools(self) -> List[Dict[str, Any]]:
"""Load tool definitions from tools.json"""
tools_file = Path(__file__).parent / "tools.json"
with open(tools_file, 'r') as f:
data = json.load(f)
return data["tools"]
def _load_system_prompt(self) -> str:
"""Load system prompt from system-prompt.md"""
prompt_file = Path(__file__).parent / "system-prompt.md"
with open(prompt_file, 'r') as f:
content = f.read()
# Inject current environment information
content = content.replace("${Working directory}", self.system_state.current_directory)
content = content.replace("${current_branch}", self._get_git_branch())
content = content.replace("${main_branch}", self._get_main_branch())
content = content.replace("${git status}", self._get_git_status())
content = content.replace("${Last 5 Recent commits}", self._get_recent_commits())
return content
def _get_git_branch(self) -> str:
"""Get current git branch"""
try:
import subprocess
return subprocess.getoutput("git branch --show-current") or "unknown"
except:
return "unknown"
def _get_main_branch(self) -> str:
"""Get main branch name"""
try:
import subprocess
branches = subprocess.getoutput("git branch -a")
if "main" in branches:
return "main"
elif "master" in branches:
return "master"
return "main"
except:
return "main"
def _get_git_status(self) -> str:
"""Get git status"""
try:
import subprocess
return subprocess.getoutput("git status --short") or "No changes"
except:
return "Not a git repository"
def _get_recent_commits(self) -> str:
"""Get recent commits"""
try:
import subprocess
return subprocess.getoutput("git log --oneline -5") or "No commits"
except:
return "Not a git repository"
def run(self, user_message: str, max_iterations: int = 50) -> Iterator[Dict[str, Any]]:
"""
Run agent with streaming output
Args:
user_message: User's input message
max_iterations: Maximum number of agent iterations
Yields:
Streaming events with type and content
"""
# Add user message with timestamp
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
self.messages.append({
"role": "user",
"content": f"[{timestamp}] {user_message}"
})
yield {"type": "user_message", "content": user_message, "timestamp": timestamp}
for iteration in range(max_iterations):
yield {"type": "iteration_start", "iteration": iteration + 1}
# Prepare messages with system hint
messages_with_hint = self.messages.copy()
# Add system hint as a user message at the end
system_hint = self.system_state.get_system_hint()
messages_with_hint.append({
"role": "user",
"content": f"<system_hint>\n{system_hint}\n</system_hint>"
})
# Call API based on client type
try:
if self.client_type == "anthropic":
# Use Anthropic streaming
iteration_generator = self._run_anthropic_iteration(messages_with_hint, iteration)
else:
# Use OpenAI/OpenRouter streaming
iteration_generator = self._run_openai_iteration(messages_with_hint, iteration)
should_break = False
for event in iteration_generator:
yield event
if event.get("type") == "done":
should_break = True
if should_break:
break
except Exception as e:
yield {"type": "error", "error": str(e)}
break
else:
# Reached max iterations
yield {"type": "max_iterations_reached", "max_iterations": max_iterations}
def _run_anthropic_iteration(self, messages_with_hint: List[Dict[str, Any]], iteration: int) -> Iterator[Dict[str, Any]]:
"""Run one iteration with Anthropic API"""
try:
with self.client.messages.stream(
model=self.model,
system=self.system_prompt,
messages=messages_with_hint,
tools=self.tools
) as stream:
assistant_message = {"role": "assistant", "content": []}
current_text = ""
current_tool_use = None
for event in stream:
if event.type == "content_block_start":
if event.content_block.type == "text":
current_text = ""
elif event.content_block.type == "tool_use":
current_tool_use = {
"type": "tool_use",
"id": event.content_block.id,
"name": event.content_block.name,
"input": {}
}
elif event.type == "content_block_delta":
if event.delta.type == "text_delta":
current_text += event.delta.text
yield {
"type": "text_delta",
"delta": event.delta.text,
"accumulated": current_text
}
elif event.delta.type == "input_json_delta":
# Accumulate tool input
if current_tool_use:
try:
current_tool_use["input"] = json.loads(
event.delta.partial_json
)
except:
pass
elif event.type == "content_block_stop":
if current_text:
assistant_message["content"].append({
"type": "text",
"text": current_text
})
current_text = ""
elif current_tool_use:
assistant_message["content"].append(current_tool_use)
yield {
"type": "tool_call",
"tool": current_tool_use["name"],
"input": current_tool_use["input"]
}
current_tool_use = None
# Add assistant message to history
self.messages.append(assistant_message)
# Check if we have tool calls to execute
tool_calls = [
block for block in assistant_message["content"]
if block.get("type") == "tool_use"
]
if not tool_calls:
# No tool calls, agent is done
yield {"type": "done", "final_message": assistant_message}
return
# Execute tool calls
tool_results = []
for tool_call in tool_calls:
tool_name = tool_call["name"]
tool_input = tool_call["input"]
yield {
"type": "tool_execution_start",
"tool": tool_name,
"input": tool_input
}
# Get tool instance and execute
try:
tool = self.tool_registry.get_tool(tool_name, self.system_state)
result = tool.execute(tool_input)
result_dict = result.to_dict()
except Exception as e:
result_dict = {
"error": str(e),
"tool": tool_name
}
yield {
"type": "tool_execution_complete",
"tool": tool_name,
"result": result_dict
}
tool_results.append({
"type": "tool_result",
"tool_use_id": tool_call["id"],
"content": json.dumps(result_dict, ensure_ascii=False)
})
# Add tool results to messages
self.messages.append({
"role": "user",
"content": tool_results
})
yield {"type": "iteration_end", "iteration": iteration + 1}
except Exception as e:
raise e
def _run_openai_iteration(self, messages_with_hint: List[Dict[str, Any]], iteration: int) -> Iterator[Dict[str, Any]]:
"""Run one iteration with OpenAI/OpenRouter API"""
try:
# Convert messages to OpenAI format
openai_messages = self._convert_to_openai_format(messages_with_hint)
# Convert tools to OpenAI format
openai_tools = self._convert_tools_to_openai_format()
# Stream completion
params = {
"model": self.model,
"messages": openai_messages,
"tools": openai_tools,
"stream": True,
}
stream = self.client.chat.completions.create(**params)
assistant_message = {"role": "assistant", "content": ""}
tool_calls_data = {}
current_text = ""
for chunk in stream:
delta = chunk.choices[0].delta if chunk.choices else None
if not delta:
continue
# Handle text content
if delta.content:
current_text += delta.content
yield {
"type": "text_delta",
"delta": delta.content,
"accumulated": current_text
}
# Handle tool calls
if delta.tool_calls:
for tool_call in delta.tool_calls:
idx = tool_call.index
if idx not in tool_calls_data:
tool_calls_data[idx] = {
"id": tool_call.id or "",
"name": "",
"arguments": ""
}
if tool_call.function.name:
tool_calls_data[idx]["name"] = tool_call.function.name
if tool_call.function.arguments:
tool_calls_data[idx]["arguments"] += tool_call.function.arguments
# Store assistant message
assistant_message["content"] = current_text
if tool_calls_data:
assistant_message["tool_calls"] = list(tool_calls_data.values())
self.messages.append(assistant_message)
# Check if we have tool calls
if not tool_calls_data:
yield {"type": "done", "final_message": assistant_message}
return
# Execute tool calls
tool_results = []
for tool_call in tool_calls_data.values():
tool_name = tool_call["name"]
try:
tool_input = json.loads(tool_call["arguments"])
except:
tool_input = {}
yield {
"type": "tool_call",
"tool": tool_name,
"input": tool_input
}
yield {
"type": "tool_execution_start",
"tool": tool_name,
"input": tool_input
}
# Execute tool
try:
tool = self.tool_registry.get_tool(tool_name, self.system_state)
result = tool.execute(tool_input)
result_dict = result.to_dict()
except Exception as e:
result_dict = {
"error": str(e),
"tool": tool_name
}
yield {
"type": "tool_execution_complete",
"tool": tool_name,
"result": result_dict
}
tool_results.append({
"role": "tool",
"tool_call_id": tool_call["id"],
"content": json.dumps(result_dict, ensure_ascii=False)
})
# Add tool results to messages
self.messages.extend(tool_results)
yield {"type": "iteration_end", "iteration": iteration + 1}
except Exception as e:
raise e
def _convert_to_openai_format(self, messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""Convert Anthropic message format to OpenAI format"""
openai_messages = []
# Add system message
openai_messages.append({
"role": "system",
"content": self.system_prompt
})
for msg in messages:
role = msg["role"]
content = msg.get("content", "")
if role == "user":
if isinstance(content, str):
openai_messages.append({"role": "user", "content": content})
elif isinstance(content, list):
# Handle tool results
for item in content:
if item.get("type") == "tool_result":
openai_messages.append({
"role": "tool",
"tool_call_id": item.get("tool_use_id", ""),
"content": item.get("content", "")
})
elif role == "assistant":
msg_dict = {"role": "assistant", "content": content}
if "tool_calls" in msg and msg["tool_calls"]:
msg_dict["tool_calls"] = [
{
"id": tc["id"],
"type": "function",
"function": {
"name": tc["name"],
"arguments": tc["arguments"]
}
} for tc in msg["tool_calls"]
]
openai_messages.append(msg_dict)
elif role == "tool":
# 保留 tool 消息
openai_messages.append({
"role": "tool",
"tool_call_id": msg.get("tool_call_id", ""),
"content": content
})
return openai_messages
def _convert_tools_to_openai_format(self) -> List[Dict[str, Any]]:
"""Convert Anthropic tool format to OpenAI format"""
openai_tools = []
for tool in self.tools:
openai_tools.append({
"type": "function",
"function": {
"name": tool["name"],
"description": tool["description"],
"parameters": tool["input_schema"]
}
})
return openai_tools
def reset(self):
"""Reset agent state"""
self.messages = []
self.system_state = SystemState()
def main():
"""Example usage"""
api_key = os.getenv("ANTHROPIC_API_KEY")
if not api_key:
print("Error: ANTHROPIC_API_KEY environment variable not set")
return
agent = CodingAgent(api_key=api_key)
user_query = "List all Python files in the current directory using the Glob tool"
print(f"User: {user_query}\n")
for event in agent.run(user_query):
if event["type"] == "text_delta":
print(event["delta"], end="", flush=True)
elif event["type"] == "tool_call":
print(f"\n[Calling tool: {event['tool']}]")
elif event["type"] == "tool_execution_complete":
print(f"[Tool completed]")
elif event["type"] == "done":
print("\n\nAgent completed successfully!")
elif event["type"] == "error":
print(f"\n\nError: {event['error']}")
if __name__ == "__main__":
main()
agent_new.py¶
"""
Comprehensive Coding Agent - Modular implementation with pure Python tools
All tools implemented without command-line dependencies
"""
import os
import json
from typing import List, Dict, Any, Optional, Iterator
from pathlib import Path
from datetime import datetime
import anthropic
from system_state import SystemState
from tool_registry import ToolRegistry
class CodingAgent:
"""Main coding agent with streaming support and modular tool system"""
def __init__(self, api_key: str, model: str = "claude-sonnet-4-20250514", base_url: Optional[str] = None):
"""
Initialize coding agent
Args:
api_key: API key for Anthropic/OpenRouter
model: Model identifier
base_url: Optional base URL for OpenRouter or other providers
"""
self.api_key = api_key
self.model = model
self.base_url = base_url
self.system_state = SystemState()
self.tool_registry = ToolRegistry()
self.messages: List[Dict[str, Any]] = []
self.tools = self._load_tools()
self.system_prompt = self._load_system_prompt()
# Initialize Anthropic client
if base_url:
self.client = anthropic.Anthropic(api_key=api_key, base_url=base_url)
else:
self.client = anthropic.Anthropic(api_key=api_key)
def _load_tools(self) -> List[Dict[str, Any]]:
"""Load tool definitions from tools.json"""
tools_file = Path(__file__).parent / "tools.json"
with open(tools_file, 'r') as f:
data = json.load(f)
return data["tools"]
def _load_system_prompt(self) -> str:
"""Load system prompt from system-prompt.md"""
prompt_file = Path(__file__).parent / "system-prompt.md"
with open(prompt_file, 'r') as f:
content = f.read()
# Inject current environment information
content = content.replace("${Working directory}", self.system_state.current_directory)
content = content.replace("${current_branch}", self._get_git_branch())
content = content.replace("${main_branch}", self._get_main_branch())
content = content.replace("${Last 5 Recent commits}", self._get_recent_commits())
return content
def _get_git_branch(self) -> str:
"""Get current git branch"""
try:
import subprocess
return subprocess.getoutput("git branch --show-current") or "unknown"
except:
return "unknown"
def _get_main_branch(self) -> str:
"""Get main branch name"""
try:
import subprocess
branches = subprocess.getoutput("git branch -a")
if "main" in branches:
return "main"
elif "master" in branches:
return "master"
return "main"
except:
return "main"
def _get_git_status(self) -> str:
"""Get git status"""
try:
import subprocess
return subprocess.getoutput("git status --short") or "No changes"
except:
return "Not a git repository"
def _get_recent_commits(self) -> str:
"""Get recent commits"""
try:
import subprocess
return subprocess.getoutput("git log --oneline -5") or "No commits"
except:
return "Not a git repository"
def run(self, user_message: str, max_iterations: int = 50) -> Iterator[Dict[str, Any]]:
"""
Run agent with streaming output
Args:
user_message: User's input message
max_iterations: Maximum number of agent iterations
Yields:
Streaming events with type and content
"""
# Add user message with timestamp
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
self.messages.append({
"role": "user",
"content": f"[{timestamp}] {user_message}"
})
yield {"type": "user_message", "content": user_message, "timestamp": timestamp}
for iteration in range(max_iterations):
yield {"type": "iteration_start", "iteration": iteration + 1}
# Prepare messages with system hint
messages_with_hint = self.messages.copy()
# Add system hint as a user message at the end
system_hint = self.system_state.get_system_hint()
messages_with_hint.append({
"role": "user",
"content": f"<system_hint>\n{system_hint}\n</system_hint>"
})
# Call Claude API with streaming
try:
with self.client.messages.stream(
model=self.model,
system=self.system_prompt,
messages=messages_with_hint,
tools=self.tools
) as stream:
assistant_message = {"role": "assistant", "content": []}
current_text = ""
current_tool_use = None
for event in stream:
if event.type == "content_block_start":
if event.content_block.type == "text":
current_text = ""
elif event.content_block.type == "tool_use":
current_tool_use = {
"type": "tool_use",
"id": event.content_block.id,
"name": event.content_block.name,
"input": {}
}
elif event.type == "content_block_delta":
if event.delta.type == "text_delta":
current_text += event.delta.text
yield {
"type": "text_delta",
"delta": event.delta.text,
"accumulated": current_text
}
elif event.delta.type == "input_json_delta":
# Accumulate tool input
if current_tool_use:
try:
current_tool_use["input"] = json.loads(
event.delta.partial_json
)
except:
pass
elif event.type == "content_block_stop":
if current_text:
assistant_message["content"].append({
"type": "text",
"text": current_text
})
current_text = ""
elif current_tool_use:
assistant_message["content"].append(current_tool_use)
yield {
"type": "tool_call",
"tool": current_tool_use["name"],
"input": current_tool_use["input"]
}
current_tool_use = None
# Add assistant message to history
self.messages.append(assistant_message)
# Check if we have tool calls to execute
tool_calls = [
block for block in assistant_message["content"]
if block.get("type") == "tool_use"
]
if not tool_calls:
# No tool calls, agent is done
yield {"type": "done", "final_message": assistant_message}
break
# Execute tool calls
tool_results = []
for tool_call in tool_calls:
tool_name = tool_call["name"]
tool_input = tool_call["input"]
yield {
"type": "tool_execution_start",
"tool": tool_name,
"input": tool_input
}
# Get tool instance and execute
try:
tool = self.tool_registry.get_tool(tool_name, self.system_state)
result = tool.execute(tool_input)
result_dict = result.to_dict()
except Exception as e:
result_dict = {
"error": str(e),
"tool": tool_name
}
yield {
"type": "tool_execution_complete",
"tool": tool_name,
"result": result_dict
}
tool_results.append({
"type": "tool_result",
"tool_use_id": tool_call["id"],
"content": json.dumps(result_dict, ensure_ascii=False)
})
# Add tool results to messages
self.messages.append({
"role": "user",
"content": tool_results
})
yield {"type": "iteration_end", "iteration": iteration + 1}
except Exception as e:
yield {"type": "error", "error": str(e)}
break
else:
# Reached max iterations
yield {"type": "max_iterations_reached", "max_iterations": max_iterations}
def reset(self):
"""Reset agent state"""
self.messages = []
self.system_state = SystemState()
def main():
"""Example usage"""
api_key = os.getenv("ANTHROPIC_API_KEY")
if not api_key:
print("Error: ANTHROPIC_API_KEY environment variable not set")
return
agent = CodingAgent(api_key=api_key)
user_query = "List all Python files in the current directory using the Glob tool"
print(f"User: {user_query}\n")
for event in agent.run(user_query):
if event["type"] == "text_delta":
print(event["delta"], end="", flush=True)
elif event["type"] == "tool_call":
print(f"\n[Calling tool: {event['tool']}]")
elif event["type"] == "tool_execution_complete":
print(f"[Tool completed]")
elif event["type"] == "done":
print("\n\nAgent completed successfully!")
elif event["type"] == "error":
print(f"\n\nError: {event['error']}")
if __name__ == "__main__":
main()
config.py¶
"""
Configuration for the coding agent
"""
import os
from pathlib import Path
from dotenv import load_dotenv
# Load environment variables
load_dotenv()
class Config:
"""Configuration class for the coding agent"""
# API Configuration
ANTHROPIC_API_KEY = os.getenv("ANTHROPIC_API_KEY")
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
OPENROUTER_API_KEY = os.getenv("OPENROUTER_API_KEY")
# Provider selection (anthropic, openai, or openrouter)
PROVIDER = os.getenv("PROVIDER", "anthropic").lower()
# Default model
DEFAULT_MODEL = os.getenv("DEFAULT_MODEL", "claude-sonnet-5")
# OpenRouter configuration
OPENROUTER_BASE_URL = "https://openrouter.ai/api/v1"
# Agent configuration
MAX_ITERATIONS = int(os.getenv("MAX_ITERATIONS", "50"))
MAX_TOKENS = int(os.getenv("MAX_TOKENS", "8192"))
# System configuration
WORKING_DIRECTORY = os.getenv("WORKING_DIRECTORY", os.getcwd())
@classmethod
def get_provider(cls) -> str:
"""Get the configured provider"""
return cls.PROVIDER
@classmethod
def get_api_key(cls, provider: str = None) -> str:
"""Get API key for specified provider (or configured provider if not specified)"""
if provider is None:
provider = cls.PROVIDER
if provider == "anthropic":
if not cls.ANTHROPIC_API_KEY:
raise ValueError("ANTHROPIC_API_KEY not set in .env file")
return cls.ANTHROPIC_API_KEY
elif provider == "openai":
if not cls.OPENAI_API_KEY:
raise ValueError("OPENAI_API_KEY not set in .env file")
return cls.OPENAI_API_KEY
elif provider == "openrouter":
if not cls.OPENROUTER_API_KEY:
raise ValueError("OPENROUTER_API_KEY not set in .env file")
return cls.OPENROUTER_API_KEY
else:
raise ValueError(f"Unknown provider: {provider}. Must be one of: anthropic, openai, openrouter")
@classmethod
def get_base_url(cls) -> str:
"""Get base URL for the configured provider"""
if cls.PROVIDER == "openrouter":
return cls.OPENROUTER_BASE_URL
return None # Use default for anthropic/openai
@staticmethod
def map_model_to_openrouter(model: str) -> str:
"""Map a native (anthropic/openai) model id to an OpenRouter model id.
Used by the OpenRouter fallback so a user with ONLY an OPENROUTER_API_KEY
can still run a task written for a direct provider.
- already-prefixed ids (contain "/") are passed through unchanged
- claude-sonnet-* -> anthropic/claude-sonnet-4.6
- claude-haiku-* -> anthropic/claude-haiku-4.5
- claude-opus-* -> anthropic/claude-opus-4.8
- any other claude-* -> anthropic/claude-opus-4.8
- gpt-* / o1-* -> openai/<model>
- anything else -> returned unchanged (best effort)
"""
if "/" in model:
return model # already an OpenRouter id
m = model.lower()
if m.startswith("claude"):
if "haiku" in m:
return "anthropic/claude-haiku-4.5"
if "sonnet" in m:
return "anthropic/claude-sonnet-4.6"
# opus, or any other/unknown Claude tier -> latest Opus
return "anthropic/claude-opus-4.8"
if m.startswith("gpt-") or m.startswith("o1"):
return f"openai/{model}"
return model
@classmethod
def resolve(cls) -> dict:
"""Resolve the effective (provider, api_key, base_url, model).
Applies the OpenRouter universal fallback: if the requested direct
provider (anthropic/openai) has no API key configured, but an
OPENROUTER_API_KEY is available, transparently route through OpenRouter
(OpenAI-compatible SDK + prefixed model id). Default behavior is
unchanged whenever the requested provider's own key is present.
Returns a dict: {provider, api_key, base_url, model,
requested_provider, fell_back}.
"""
requested = cls.PROVIDER
model = cls.DEFAULT_MODEL
# Explicit OpenRouter selection: use as configured (no mapping).
if requested == "openrouter":
return {
"provider": "openrouter",
"api_key": cls.get_api_key("openrouter"),
"base_url": cls.OPENROUTER_BASE_URL,
"model": model,
"requested_provider": requested,
"fell_back": False,
}
if requested in ("anthropic", "openai"):
direct_key = cls.ANTHROPIC_API_KEY if requested == "anthropic" else cls.OPENAI_API_KEY
if direct_key:
# Direct provider key present -> behave exactly as before.
return {
"provider": requested,
"api_key": direct_key,
"base_url": None,
"model": model,
"requested_provider": requested,
"fell_back": False,
}
# No direct key -> fall back to OpenRouter if we have that key.
if cls.OPENROUTER_API_KEY:
return {
"provider": "openrouter",
"api_key": cls.OPENROUTER_API_KEY,
"base_url": cls.OPENROUTER_BASE_URL,
"model": cls.map_model_to_openrouter(model),
"requested_provider": requested,
"fell_back": True,
}
key_name = "ANTHROPIC_API_KEY" if requested == "anthropic" else "OPENAI_API_KEY"
raise ValueError(
f"No {key_name} set and no OPENROUTER_API_KEY available for fallback. "
f"Set {key_name} for direct access, or set OPENROUTER_API_KEY to route "
f"'{requested}' models through OpenRouter."
)
raise ValueError(f"Unknown provider: {requested}. Must be one of: anthropic, openai, openrouter")
@classmethod
def validate(cls):
"""Validate configuration (fallback-aware)."""
# resolve() raises a clear error if neither the direct key nor the
# OpenRouter fallback key is available.
resolved = cls.resolve()
provider = cls.get_provider()
# Validate model name only when using the direct provider path; when we
# fall back to OpenRouter the model id is remapped/prefixed, so the
# native naming rules no longer apply.
if not resolved["fell_back"]:
if provider == "anthropic" and not cls.DEFAULT_MODEL.startswith("claude"):
raise ValueError(f"Model '{cls.DEFAULT_MODEL}' is not valid for Anthropic. Use a model starting with 'claude-'")
elif provider == "openai" and not any(cls.DEFAULT_MODEL.startswith(p) for p in ["gpt-", "o1-"]):
raise ValueError(f"Model '{cls.DEFAULT_MODEL}' is not valid for OpenAI. Use a model starting with 'gpt-' or 'o1-'")
# OpenRouter accepts any model name, so no validation needed
example_complex_task.py¶
#!/usr/bin/env python3
"""
Example of a complex coding task with the agent
"""
from agent import CodingAgent
from config import Config
def run_complex_task():
"""Run a complex multi-step task"""
Config.validate()
provider = Config.get_provider()
api_key = Config.get_api_key()
base_url = Config.get_base_url()
agent = CodingAgent(
api_key=api_key,
model=Config.DEFAULT_MODEL,
base_url=base_url,
provider=provider
)
# Complex task that requires multiple steps
user_query = """
I need you to create a simple web scraper project:
1. Create a directory called 'web_scraper'
2. Inside it, create:
- requirements.txt with requests and beautifulsoup4
- scraper.py with a function that:
* Takes a URL as input
* Fetches the page
* Extracts all links
* Returns them as a list
- test_scraper.py with basic unit tests
- README.md with usage instructions
3. After creating everything, check if there are any syntax errors in the Python files
Please implement this step by step, using the TODO list to track your progress.
"""
print("=" * 80)
print("COMPLEX CODING TASK EXAMPLE")
print("=" * 80)
print(f"\nUser: {user_query.strip()}\n")
print("-" * 80)
# Track statistics
tool_calls = 0
iterations = 0
for event in agent.run(user_query, max_iterations=30):
if event["type"] == "iteration_start":
iterations = event["iteration"]
print(f"\n[Iteration {iterations}]")
elif event["type"] == "text_delta":
print(event["delta"], end="", flush=True)
elif event["type"] == "tool_call":
tool_calls += 1
print(f"\n\n🔧 Tool #{tool_calls}: {event['tool']}")
elif event["type"] == "tool_execution_complete":
result = event["result"]
# Show TODO updates
if event["tool"] == "TodoWrite":
print(f" ✓ TODO list updated:")
print(f" - Total: {result.get('total_todos', 0)}")
print(f" - In Progress: {result.get('in_progress', 0)}")
print(f" - Completed: {result.get('completed', 0)}")
# Show file operations
elif event["tool"] in ["Write", "Edit", "MultiEdit"]:
print(f" ✓ File: {result.get('file_path', 'unknown')}")
if "lint_check" in result:
lint = result["lint_check"]
if lint.get("has_errors"):
print(f" ⚠️ Lint errors detected!")
else:
print(f" ✓ No lint errors")
# Show bash results
elif event["tool"] == "Bash":
exit_code = result.get("exit_code", -1)
if exit_code == 0:
print(f" ✓ Command executed successfully")
else:
print(f" ⚠️ Command failed with exit code {exit_code}")
elif event["type"] == "done":
print("\n\n" + "=" * 80)
print("✅ TASK COMPLETED!")
print(f" Total iterations: {iterations}")
print(f" Total tool calls: {tool_calls}")
print("=" * 80)
elif event["type"] == "error":
print(f"\n\n❌ Error: {event['error']}")
elif event["type"] == "max_iterations_reached":
print(f"\n\n⚠️ Reached maximum iterations")
if __name__ == "__main__":
run_complex_task()
example_with_system_hints.py¶
#!/usr/bin/env python3
"""
Example demonstrating system hints features
"""
from agent import CodingAgent
from config import Config
def demo_system_hints():
"""Demonstrate system hint features like timestamps, tool counting, and TODO tracking"""
Config.validate()
provider = Config.get_provider()
api_key = Config.get_api_key()
base_url = Config.get_base_url()
agent = CodingAgent(
api_key=api_key,
model=Config.DEFAULT_MODEL,
base_url=base_url,
provider=provider
)
user_query = """
Let's test the system hint features:
1. Create 3 simple Python files (test1.py, test2.py, test3.py)
2. Each should just print a different message
3. Use a TODO list to track your progress
4. After creating them, read each one back to verify
This will help demonstrate:
- Timestamps on each operation
- Tool call counting
- TODO list tracking
- Working directory persistence across commands
"""
print("=" * 80)
print("SYSTEM HINTS DEMONSTRATION")
print("=" * 80)
print("\nThis example will show:")
print(" • Timestamps on all operations")
print(" • Tool call counters (watch for warnings after 3+ calls)")
print(" • TODO list tracking")
print(" • Persistent shell session")
print(" • Automatic lint checking")
print("=" * 80)
print(f"\nUser: {user_query.strip()}\n")
print("-" * 80)
for event in agent.run(user_query, max_iterations=30):
if event["type"] == "user_message":
print(f"\n[{event['timestamp']}] User message received")
elif event["type"] == "iteration_start":
print(f"\n{'='*60}")
print(f"ITERATION {event['iteration']}")
# Show current system state
state = agent.system_state
print(f"System State:")
print(f" Working Directory: {state.current_directory}")
print(f" Tool Calls: {sum(state.tool_call_counts.values())}")
if state.todos:
print(f" TODOs: {len(state.todos)} total")
print('='*60)
elif event["type"] == "text_delta":
print(event["delta"], end="", flush=True)
elif event["type"] == "tool_call":
print(f"\n\n🔧 Calling: {event['tool']}")
elif event["type"] == "tool_execution_complete":
result = event["result"]
metadata = result.get("_metadata", {})
print(f" [{metadata.get('timestamp')}] Call #{metadata.get('call_number')}")
# Highlight repeated tool calls
if metadata.get('call_number', 0) >= 3:
print(f" ⚠️ This tool has been called {metadata.get('call_number')} times!")
# Show TODO updates
if event["tool"] == "TodoWrite":
print(f" 📋 TODO List:")
print(f" Pending: {result.get('pending', 0)}")
print(f" In Progress: {result.get('in_progress', 0)}")
print(f" Completed: {result.get('completed', 0)}")
# Show actual TODOs
if agent.system_state.todos:
for todo in agent.system_state.todos:
status = todo['status']
icon = {"pending": "⬜", "in_progress": "🔄", "completed": "✅"}[status]
print(f" {icon} {todo['content']}")
elif event["tool"] == "Write":
print(f" 📝 Created: {result.get('file_path')}")
if "lint_check" in result:
lint = result["lint_check"]
if lint.get("has_errors"):
print(f" ❌ Lint errors found!")
else:
print(f" ✅ No syntax errors")
elif event["tool"] == "Bash":
wd = result.get("working_directory", "unknown")
print(f" 💻 Working directory: {wd}")
print(f" Exit code: {result.get('exit_code', 'unknown')}")
elif event["type"] == "done":
print("\n\n" + "=" * 80)
print("✅ DEMONSTRATION COMPLETE!")
print("\nFinal System State:")
state = agent.system_state
print(f" • Total tool calls: {sum(state.tool_call_counts.values())}")
print(f" • Tool breakdown:")
for tool, count in sorted(state.tool_call_counts.items()):
print(f" - {tool}: {count}")
if state.todos:
completed = sum(1 for t in state.todos if t['status'] == 'completed')
print(f" • TODOs: {completed}/{len(state.todos)} completed")
print("=" * 80)
elif event["type"] == "error":
print(f"\n\n❌ Error: {event['error']}")
if __name__ == "__main__":
demo_system_hints()
hello_world.py¶
#!/usr/bin/env python3
from typing import Any
def greet(name: str) -> str:
"""Return a friendly greeting for the given name."""
return f"Hello, {name}!"
def main() -> None:
# Requirement 1: Print "Hello, World!"
print("Hello, World!")
# Requirement 3: Demonstrate the function
print(greet("Alice"))
if __name__ == "__main__":
main()
main.py¶
#!/usr/bin/env python3
"""
Interactive CLI for the Coding Agent
Provides a command-line interface for chatting with the agent
"""
import os
import sys
from pathlib import Path
from agent import CodingAgent
from config import Config
class Colors:
"""ANSI color codes for terminal output"""
RESET = '\033[0m'
BOLD = '\033[1m'
DIM = '\033[2m'
# Foreground colors
BLACK = '\033[30m'
RED = '\033[31m'
GREEN = '\033[32m'
YELLOW = '\033[33m'
BLUE = '\033[34m'
MAGENTA = '\033[35m'
CYAN = '\033[36m'
WHITE = '\033[37m'
# Bright foreground colors
BRIGHT_BLACK = '\033[90m'
BRIGHT_RED = '\033[91m'
BRIGHT_GREEN = '\033[92m'
BRIGHT_YELLOW = '\033[93m'
BRIGHT_BLUE = '\033[94m'
BRIGHT_MAGENTA = '\033[95m'
BRIGHT_CYAN = '\033[96m'
BRIGHT_WHITE = '\033[97m'
class CodingAgentCLI:
"""Interactive CLI for the Coding Agent"""
def __init__(self, use_colors: bool = True):
self.use_colors = use_colors
self.agent = None
self.running = True
def color(self, text: str, color_code: str) -> str:
"""Apply color to text if colors are enabled"""
if self.use_colors:
return f"{color_code}{text}{Colors.RESET}"
return text
def print_header(self):
"""Print CLI header"""
print()
print(self.color("=" * 80, Colors.CYAN))
print(self.color("🤖 CODING AGENT - Interactive CLI", Colors.BOLD + Colors.CYAN))
print(self.color("=" * 80, Colors.CYAN))
print()
print(self.color("Commands:", Colors.YELLOW))
print(self.color(" /help", Colors.BRIGHT_BLACK) + " - Show this help message")
print(self.color(" /quit", Colors.BRIGHT_BLACK) + " - Exit the CLI")
print(self.color(" /exit", Colors.BRIGHT_BLACK) + " - Exit the CLI")
print(self.color(" /reset", Colors.BRIGHT_BLACK) + " - Reset the agent (clear conversation history)")
print(self.color(" /clear", Colors.BRIGHT_BLACK) + " - Clear the screen")
print(self.color(" /status", Colors.BRIGHT_BLACK) + " - Show agent status")
print()
print(self.color("Type your message and press Enter. Use Ctrl+C to interrupt.", Colors.DIM))
print(self.color("-" * 80, Colors.CYAN))
print()
def print_status(self):
"""Print agent status"""
if not self.agent:
print(self.color("❌ Agent not initialized", Colors.RED))
return
state = self.agent.system_state
print()
print(self.color("📊 Agent Status:", Colors.CYAN))
print(self.color("━" * 40, Colors.CYAN))
print(f" Model: {self.color(self.agent.model, Colors.GREEN)}")
print(f" Working Directory: {self.color(state.current_directory, Colors.BLUE)}")
print(f" OS: {self.color(state.os_type, Colors.BLUE)}")
print(f" Python: {self.color(state.python_version, Colors.BLUE)}")
print(f" Messages in History: {self.color(str(len(self.agent.messages)), Colors.YELLOW)}")
if state.tool_call_counts:
print(f"\n {self.color('Tool Calls:', Colors.MAGENTA)}")
for tool, count in sorted(state.tool_call_counts.items()):
print(f" • {tool}: {self.color(str(count), Colors.YELLOW)}")
if state.todos:
print(f"\n {self.color('TODO List:', Colors.MAGENTA)}")
for todo in state.todos:
status_icons = {
"pending": "⬜",
"in_progress": "🔄",
"completed": "✅"
}
icon = status_icons.get(todo['status'], '?')
status_color = {
"pending": Colors.BRIGHT_BLACK,
"in_progress": Colors.YELLOW,
"completed": Colors.GREEN
}.get(todo['status'], Colors.WHITE)
print(f" {icon} [{todo['id']}] {self.color(todo['content'], status_color)}")
print(self.color("━" * 40, Colors.CYAN))
print()
def initialize_agent(self, model: str = None, provider: str = None, base_url: str = None):
"""Initialize the agent.
Optional overrides (model/provider/base_url) take precedence over the
values in the .env file; anything left as None falls back to Config.
"""
try:
# Apply command-line overrides on top of the .env configuration
if provider:
Config.PROVIDER = provider.lower()
if model:
Config.DEFAULT_MODEL = model
Config.validate()
# Resolve the effective provider/key/model, applying the OpenRouter
# universal fallback when a direct-provider key is missing.
resolved = Config.resolve()
provider = resolved["provider"]
api_key = resolved["api_key"]
model = resolved["model"]
# An explicit --base-url wins; otherwise use the resolved base URL
base_url = base_url if base_url else resolved["base_url"]
# Initialize agent
self.agent = CodingAgent(
api_key=api_key,
model=model,
base_url=base_url,
provider=provider
)
print(self.color("✓ Agent initialized successfully", Colors.GREEN))
if resolved["fell_back"]:
print(self.color(
f" ⚠️ No {resolved['requested_provider'].upper()} key found — "
f"falling back to OpenRouter", Colors.YELLOW))
print(self.color(
f" Requested provider: {resolved['requested_provider']} "
f"(model '{Config.DEFAULT_MODEL}')", Colors.DIM))
print(self.color(f" Provider: {provider}", Colors.DIM))
print(self.color(f" Model: {model}", Colors.DIM))
if base_url:
print(self.color(f" Base URL: {base_url}", Colors.DIM))
print()
except Exception as e:
print(self.color(f"❌ Failed to initialize agent: {str(e)}", Colors.RED))
print()
print(self.color("Please check your .env file configuration:", Colors.YELLOW))
print(self.color("Example:", Colors.DIM))
print(self.color(" PROVIDER=anthropic", Colors.DIM))
print(self.color(" ANTHROPIC_API_KEY=sk-ant-api03-...", Colors.DIM))
print(self.color(" DEFAULT_MODEL=claude-sonnet-5", Colors.DIM))
print()
print(self.color("Supported providers: anthropic, openai, openrouter", Colors.DIM))
print()
sys.exit(1)
def handle_command(self, command: str) -> bool:
"""Handle special commands. Returns True if it was a command, False otherwise."""
command = command.strip().lower()
if command in ['/quit', '/exit']:
print()
print(self.color("👋 Goodbye!", Colors.CYAN))
print()
self.running = False
return True
elif command == '/help':
self.print_header()
return True
elif command == '/reset':
if self.agent:
self.agent.reset()
print()
print(self.color("✓ Agent reset - conversation history cleared", Colors.GREEN))
print()
return True
elif command == '/clear':
os.system('clear' if os.name != 'nt' else 'cls')
self.print_header()
return True
elif command == '/status':
self.print_status()
return True
return False
def run_agent(self, user_input: str, max_iterations: int = 50):
"""Run the agent with user input and display results"""
print()
print(self.color("━" * 80, Colors.BRIGHT_BLACK))
iteration_count = 0
tool_call_count = 0
try:
for event in self.agent.run(user_input, max_iterations=max_iterations):
if event["type"] == "iteration_start":
iteration_count = event["iteration"]
if iteration_count > 1:
print()
print(self.color(f"[Iteration {iteration_count}]", Colors.DIM))
elif event["type"] == "text_delta":
# Print streaming text
print(event["delta"], end="", flush=True)
elif event["type"] == "tool_call":
tool_call_count += 1
tool_name = event["tool"]
print(f"\n\n{self.color('🔧', Colors.CYAN)} {self.color(f'Calling tool:', Colors.CYAN)} {self.color(tool_name, Colors.BOLD + Colors.YELLOW)}")
# Show tool input (abbreviated)
tool_input = event["input"]
if len(str(tool_input)) > 100:
input_preview = str(tool_input)[:100] + "..."
else:
input_preview = str(tool_input)
print(self.color(f" Input: {input_preview}", Colors.DIM))
elif event["type"] == "tool_execution_complete":
result = event["result"]
metadata = result.get("_metadata", {})
call_num = metadata.get("call_number", "?")
# Show completion status
if "error" in result:
print(self.color(f" ✗ Error: {result['error']}", Colors.RED))
else:
print(self.color(f" ✓ Completed (call #{call_num})", Colors.GREEN))
# Show important results
if "output" in result and event["tool"] == "Bash":
output = result["output"]
if len(output) > 200:
output = output[:200] + "..."
if output.strip():
print(self.color(f" Output:", Colors.DIM))
for line in output.split('\n')[:5]:
print(self.color(f" {line}", Colors.BRIGHT_BLACK))
# Show lint check results
if "lint_check" in result:
lint = result["lint_check"]
if lint.get("has_errors"):
print(self.color(f" ⚠️ Lint errors detected!", Colors.YELLOW))
else:
print(self.color(f" ✓ No lint errors", Colors.GREEN))
# Show file operations
if "file_path" in result:
file_path = result["file_path"]
# Shorten path if too long
if len(file_path) > 50:
file_path = "..." + file_path[-47:]
print(self.color(f" File: {file_path}", Colors.BLUE))
elif event["type"] == "done":
print()
print(self.color("━" * 80, Colors.BRIGHT_BLACK))
print()
print(self.color(f"✅ Task completed!", Colors.GREEN))
print(self.color(f" Iterations: {iteration_count}", Colors.DIM))
print(self.color(f" Tool calls: {tool_call_count}", Colors.DIM))
print()
elif event["type"] == "error":
print()
print(self.color("━" * 80, Colors.BRIGHT_BLACK))
print()
print(self.color(f"❌ Error: {event['error']}", Colors.RED))
print()
elif event["type"] == "max_iterations_reached":
print()
print(self.color("━" * 80, Colors.BRIGHT_BLACK))
print()
print(self.color(f"⚠️ Reached maximum iterations ({event['max_iterations']})", Colors.YELLOW))
print()
except KeyboardInterrupt:
print()
print()
print(self.color("⚠️ Interrupted by user", Colors.YELLOW))
print()
except Exception as e:
print()
print(self.color(f"❌ Unexpected error: {str(e)}", Colors.RED))
print()
def get_user_input(self) -> str:
"""Get user input with a nice prompt"""
try:
prompt = self.color("You: ", Colors.BOLD + Colors.GREEN)
return input(prompt).strip()
except EOFError:
return "/quit"
except KeyboardInterrupt:
print()
return "/quit"
def run_once(self, prompt: str, max_iterations: int = 50,
model: str = None, provider: str = None, base_url: str = None) -> int:
"""Run a single task non-interactively and exit.
Returns a process exit code (0 = success).
"""
if not sys.stdout.isatty() or os.getenv('NO_COLOR'):
self.use_colors = False
self.initialize_agent(model=model, provider=provider, base_url=base_url)
print(self.color("You: ", Colors.BOLD + Colors.GREEN) + prompt)
self.run_agent(prompt, max_iterations=max_iterations)
return 0
def run(self, max_iterations: int = 50,
model: str = None, provider: str = None, base_url: str = None):
"""Main CLI loop"""
# Check if colors are supported
if not sys.stdout.isatty() or os.getenv('NO_COLOR'):
self.use_colors = False
# Print header
self.print_header()
# Initialize agent
self.initialize_agent(model=model, provider=provider, base_url=base_url)
# Main loop
while self.running:
try:
# Get user input
user_input = self.get_user_input()
# Skip empty input
if not user_input:
continue
# Handle commands
if user_input.startswith('/'):
self.handle_command(user_input)
continue
# Run agent
self.run_agent(user_input, max_iterations=max_iterations)
except KeyboardInterrupt:
print()
print()
confirm = input(self.color("Are you sure you want to quit? (y/n): ", Colors.YELLOW))
if confirm.lower() in ['y', 'yes']:
print()
print(self.color("👋 Goodbye!", Colors.CYAN))
print()
break
else:
print()
continue
except Exception as e:
print()
print(self.color(f"❌ Unexpected error: {str(e)}", Colors.RED))
print()
def list_tools():
"""离线打印所有已注册工具及其简介（无需 API Key）。"""
import json
tools_file = Path(__file__).parent / "tools.json"
with open(tools_file, "r", encoding="utf-8") as f:
tools = json.load(f)["tools"]
print(f"共 {len(tools)} 个工具：\n")
for tool in tools:
name = tool.get("name", "?")
desc = (tool.get("description") or "").strip().splitlines()
summary = desc[0] if desc else ""
if len(summary) > 90:
summary = summary[:90] + "..."
print(f" {name:<14} {summary}")
print()
def build_parser() -> "argparse.ArgumentParser":
import argparse
parser = argparse.ArgumentParser(
prog="python main.py",
description=(
"Coding Agent —— 一个具备完整工具集（文件读写、纯 Python Grep/Glob、"
"持久化 Shell、TodoWrite 规划等）的编码智能体。\n"
"默认进入交互式对话；也可用 -p 传入单个任务后一次性执行并退出。"
),
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog=(
"示例：\n"
" # 交互式对话（默认）\n"
" python main.py\n\n"
" # 一次性执行单个任务，完成后退出（适合脚本/CI）\n"
" python main.py -p \"用 Glob 工具列出当前目录下所有 Python 文件\"\n\n"
" # 离线查看全部可用工具（无需 API Key）\n"
" python main.py --list-tools\n\n"
" # 临时指定模型 / 供应商（覆盖 .env）\n"
" python main.py --provider openrouter --model anthropic/claude-sonnet-4\n\n"
"配置：复制 .env.example 为 .env，填入所选供应商的 API Key。"
"详见 README.md 与 PROVIDERS.md。"
),
)
parser.add_argument(
"-p", "--prompt",
metavar="任务",
help="以非交互模式运行：执行给定的单个任务后退出。省略则进入交互式对话。",
)
parser.add_argument(
"--list-tools",
action="store_true",
help="离线列出全部已注册工具及简介后退出（无需 API Key，可用于快速自检）。",
)
parser.add_argument(
"--provider",
choices=["anthropic", "openai", "openrouter"],
help="临时覆盖 .env 中的 PROVIDER 设置。",
)
parser.add_argument(
"--model",
metavar="模型名",
help="临时覆盖 .env 中的 DEFAULT_MODEL（例如 claude-sonnet-5）。",
)
parser.add_argument(
"--base-url",
metavar="URL",
help="临时覆盖 API Base URL（用于自建网关或兼容 OpenAI 的第三方服务）。",
)
parser.add_argument(
"--max-iterations",
type=int,
default=Config.MAX_ITERATIONS,
metavar="N",
help=f"单个任务的最大 Agent 迭代轮数（默认 {Config.MAX_ITERATIONS}）。",
)
parser.add_argument(
"--no-color",
action="store_true",
help="禁用彩色输出（管道 / 无 TTY 环境会自动禁用）。",
)
return parser
def main():
"""Entry point"""
parser = build_parser()
args = parser.parse_args()
# 离线路径：仅列出工具，无需初始化 Agent 或 API Key
if args.list_tools:
list_tools()
return
cli = CodingAgentCLI(use_colors=not args.no_color)
if args.prompt:
# 非交互（一次性）模式
exit_code = cli.run_once(
args.prompt,
max_iterations=args.max_iterations,
model=args.model,
provider=args.provider,
base_url=args.base_url,
)
sys.exit(exit_code)
else:
# 交互模式（默认行为，保持不变）
cli.run(
max_iterations=args.max_iterations,
model=args.model,
provider=args.provider,
base_url=args.base_url,
)
if __name__ == "__main__":
main()
quickstart.py¶
#!/usr/bin/env python3
"""
Quickstart script for the coding agent
"""
import os
from agent import CodingAgent
from config import Config
def main():
"""Run a simple coding agent interaction"""
# Validate configuration
try:
Config.validate()
except ValueError as e:
print(f"Configuration error: {e}")
print("Please set at least one API key in your .env file")
return
# Initialize agent
provider = Config.get_provider()
api_key = Config.get_api_key()
base_url = Config.get_base_url()
agent = CodingAgent(
api_key=api_key,
model=Config.DEFAULT_MODEL,
base_url=base_url,
provider=provider
)
# Example query
user_query = """
Create a simple Python script called hello_world.py that:
1. Prints "Hello, World!"
2. Has a function that greets a person by name
3. Has a main block that demonstrates the function
After creating it, run it to verify it works.
"""
print("=" * 80)
print("CODING AGENT QUICKSTART")
print("=" * 80)
print(f"\nUser: {user_query.strip()}\n")
print("-" * 80)
print()
# Run agent
for event in agent.run(user_query):
if event["type"] == "text_delta":
print(event["delta"], end="", flush=True)
elif event["type"] == "tool_call":
print(f"\n\n🔧 Calling tool: {event['tool']}")
print(f" Input: {event['input']}")
elif event["type"] == "tool_execution_complete":
result = event["result"]
metadata = result.get("_metadata", {})
print(f" ✓ {metadata.get('tool')} call #{metadata.get('call_number')} completed")
# Show important results
if "error" in result:
print(f" ⚠️ Error: {result['error']}")
elif "output" in result:
output = result["output"][:200]
print(f" Output: {output}...")
# Show lint check results
if "lint_check" in result:
lint = result["lint_check"]
if lint.get("has_errors"):
print(f" ⚠️ Lint errors found: {lint.get('errors')}")
else:
print(f" ✓ No lint errors detected")
elif event["type"] == "done":
print("\n\n" + "=" * 80)
print("✅ Agent completed successfully!")
print("=" * 80)
elif event["type"] == "error":
print(f"\n\n❌ Error: {event['error']}")
elif event["type"] == "max_iterations_reached":
print(f"\n\n⚠️ Reached maximum iterations ({event['max_iterations']})")
if __name__ == "__main__":
main()
system_state.py¶
"""
System state tracking for the coding agent
"""
import os
import subprocess
from dataclasses import dataclass, field
from typing import Dict, Any, List
from datetime import datetime
@dataclass
class SystemState:
"""System state tracking for system hints"""
current_directory: str = field(default_factory=lambda: os.getcwd())
tool_call_counts: Dict[str, int] = field(default_factory=dict)
todos: List[Dict[str, Any]] = field(default_factory=list)
shell_sessions: Dict[str, Any] = field(default_factory=dict)
default_shell_id: str = "default"
os_type: str = field(default_factory=lambda: os.uname().sysname)
python_version: str = field(default_factory=lambda: subprocess.getoutput("python3 --version"))
def get_system_hint(self) -> str:
"""Generate system hint message to append to context"""
hint_parts = []
# Environment information
hint_parts.append("# System State")
hint_parts.append(f"Current Time: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
hint_parts.append(f"Working Directory: {self.current_directory}")
hint_parts.append(f"OS: {self.os_type}")
hint_parts.append(f"Python: {self.python_version}")
# Tool call statistics
if self.tool_call_counts:
hint_parts.append("\n# Tool Call Statistics")
for tool, count in sorted(self.tool_call_counts.items()):
hint_parts.append(f"- {tool}: {count} calls")
if count >= 3:
hint_parts.append(f" ⚠️ Tool '{tool}' has been called {count} times. Consider alternative approaches.")
# TODO list
if self.todos:
hint_parts.append("\n# Current TODO List")
for todo in self.todos:
status_icon = {"pending": "⬜", "in_progress": "🔄", "completed": "✅"}[todo["status"]]
hint_parts.append(f"{status_icon} [{todo['id']}] {todo['content']} ({todo['status']})")
return "\n".join(hint_parts)
tests/__init__.py¶
tests/conftest.py¶
"""
Pytest configuration and fixtures
"""
import pytest
import tempfile
import shutil
from pathlib import Path
from system_state import SystemState
@pytest.fixture
def system_state():
"""Create a fresh system state for each test"""
return SystemState()
@pytest.fixture
def temp_dir():
"""Create a temporary directory for tests"""
temp_path = Path(tempfile.mkdtemp())
yield temp_path
# Cleanup after test
shutil.rmtree(temp_path, ignore_errors=True)
@pytest.fixture
def sample_files(temp_dir):
"""Create sample files for testing"""
# Create Python file
python_file = temp_dir / "sample.py"
python_file.write_text("""
def hello(name):
return f"Hello, {name}!"
def add(a, b):
return a + b
if __name__ == "__main__":
print(hello("World"))
""")
# Create JavaScript file
js_file = temp_dir / "sample.js"
js_file.write_text("""
function hello(name) {
return `Hello, ${name}!`;
}
function add(a, b) {
return a + b;
}
console.log(hello("World"));
""")
# Create text files
text_file1 = temp_dir / "file1.txt"
text_file1.write_text("This is a test file.\nIt has multiple lines.\nSome contain the word ERROR.\n")
text_file2 = temp_dir / "file2.txt"
text_file2.write_text("Another file here.\nNo errors in this one.\nJust normal text.\n")
# Create nested directory
nested_dir = temp_dir / "subdir"
nested_dir.mkdir()
nested_file = nested_dir / "nested.py"
nested_file.write_text("""
class TestClass:
def method(self):
pass
""")
return {
"python_file": python_file,
"js_file": js_file,
"text_file1": text_file1,
"text_file2": text_file2,
"nested_file": nested_file,
"temp_dir": temp_dir
}
tests/test_bash_output_tool.py¶
"""
Test cases for BashOutput tool
Tests all features from tools.json
"""
import pytest
import time
from pathlib import Path
from tools.bash_output_tool import BashOutputTool
from tools.bash_tool import BashTool
class TestBashOutputTool:
"""Test BashOutput tool functionality"""
def test_retrieve_background_output(self, system_state):
"""Test retrieving output from background job"""
bash_tool = BashTool(system_state)
output_tool = BashOutputTool(system_state)
# Start background job
bash_result = bash_tool.execute({
"command": "echo 'background output' && sleep 1",
"run_in_background": True
})
assert bash_result.success
bg_id = bash_result.data["background_job_id"]
# Wait a bit for output
time.sleep(0.5)
# Retrieve output
result = output_tool.execute({
"bash_id": bg_id
})
assert result.success
assert "background output" in result.data["output"]
def test_filter_parameter(self, system_state):
"""Test optional regex filtering of output"""
bash_tool = BashTool(system_state)
output_tool = BashOutputTool(system_state)
# Create background job with mixed output
bash_result = bash_tool.execute({
"command": "echo 'ERROR: something' && echo 'INFO: other' && echo 'ERROR: again'",
"run_in_background": True
})
bg_id = bash_result.data["background_job_id"]
time.sleep(0.5)
# Filter for ERROR lines only
result = output_tool.execute({
"bash_id": bg_id,
"filter": "ERROR"
})
assert result.success
output_lines = result.data["output"].split('\n')
# Should only have ERROR lines
assert all("ERROR" in line or not line.strip() for line in output_lines if line.strip())
def test_nonexistent_bash_id(self, system_state):
"""Test error when bash_id doesn't exist"""
tool = BashOutputTool(system_state)
result = tool.execute({
"bash_id": "nonexistent_12345"
})
assert "error" in result.data
assert "not found" in result.data["error"].lower()
def test_output_size_tracking(self, system_state):
"""Test that output_size is included in result"""
bash_tool = BashTool(system_state)
output_tool = BashOutputTool(system_state)
bash_result = bash_tool.execute({
"command": "echo 'test output'",
"run_in_background": True
})
bg_id = bash_result.data["background_job_id"]
time.sleep(0.5)
result = output_tool.execute({
"bash_id": bg_id
})
assert result.success
assert "output_size" in result.data
assert result.data["output_size"] > 0
tests/test_bash_tool.py¶
"""
Test cases for Bash tool
Tests all features from tools.json
"""
import pytest
import time
from pathlib import Path
from tools.bash_tool import BashTool
class TestBashTool:
"""Test Bash tool functionality"""
def test_basic_command(self, system_state):
"""Test basic command execution"""
tool = BashTool(system_state)
result = tool.execute({
"command": "echo 'Hello, World!'"
})
assert result.success
assert "Hello, World!" in result.data["output"]
assert result.data["exit_code"] == 0
def test_command_with_exit_code(self, system_state):
"""Test that exit codes are captured"""
tool = BashTool(system_state)
# Successful command
result = tool.execute({
"command": "true"
})
assert result.data["exit_code"] == 0
# Failed command
result = tool.execute({
"command": "false"
})
assert result.data["exit_code"] == 1
def test_persistent_shell_session(self, system_state, temp_dir):
"""Test that shell session persists across commands"""
tool = BashTool(system_state)
# Set an environment variable
result1 = tool.execute({
"command": "export TEST_VAR=hello"
})
assert result1.success
# Check that it persists
result2 = tool.execute({
"command": "echo $TEST_VAR"
})
assert result2.success
assert "hello" in result2.data["output"]
def test_directory_change_persistence(self, system_state, temp_dir):
"""Test that directory changes persist"""
tool = BashTool(system_state)
# Change directory
result1 = tool.execute({
"command": f"cd {temp_dir}"
})
assert result1.success
# Verify we're in the new directory
result2 = tool.execute({
"command": "pwd"
})
assert result2.success
assert str(temp_dir) in result2.data["output"]
# System state should also be updated
assert temp_dir in Path(system_state.current_directory).parents or \
Path(temp_dir) == Path(system_state.current_directory)
def test_timeout_parameter(self, system_state):
"""Test timeout parameter (in milliseconds)"""
tool = BashTool(system_state)
# Command that should timeout (1 second timeout)
result = tool.execute({
"command": "sleep 5",
"timeout": 1000 # 1 second in ms
})
assert "timeout" in result.data["output"].lower()
def test_output_truncation(self, system_state):
"""Test that output exceeding 30000 chars is truncated"""
tool = BashTool(system_state)
# Generate large output
result = tool.execute({
"command": "yes | head -n 2000"
})
assert result.success
output_len = len(result.data["output"])
# Should be truncated or close to limit
assert output_len <= 35000 # Some buffer
def test_background_execution(self, system_state):
"""Test run_in_background parameter"""
tool = BashTool(system_state)
result = tool.execute({
"command": "sleep 1 && echo done",
"run_in_background": True
})
assert result.success
assert "background_job_id" in result.data
assert "PID" in result.data["output"]
def test_multiple_commands_with_semicolon(self, system_state, temp_dir):
"""Test multiple commands separated by semicolon"""
tool = BashTool(system_state)
result = tool.execute({
"command": f"cd {temp_dir} ; touch test_file.txt ; ls test_file.txt"
})
assert result.success
assert "test_file.txt" in result.data["output"]
def test_multiple_commands_with_and(self, system_state, temp_dir):
"""Test multiple commands with && operator"""
tool = BashTool(system_state)
result = tool.execute({
"command": f"cd {temp_dir} && echo 'success'"
})
assert result.success
assert "success" in result.data["output"]
def test_quoted_paths_with_spaces(self, system_state, temp_dir):
"""Test handling paths with spaces using quotes"""
tool = BashTool(system_state)
# Create directory with spaces
space_dir = temp_dir / "dir with spaces"
space_dir.mkdir()
result = tool.execute({
"command": f'cd "{space_dir}" && pwd'
})
assert result.success
assert "dir with spaces" in result.data["output"]
def test_shell_id_tracking(self, system_state):
"""Test that shell_id is returned"""
tool = BashTool(system_state)
result = tool.execute({
"command": "echo test"
})
assert result.success
assert "shell_id" in result.data
assert result.data["shell_id"] == "default"
def test_working_directory_in_result(self, system_state):
"""Test that working_directory is included in result"""
tool = BashTool(system_state)
result = tool.execute({
"command": "pwd"
})
assert result.success
assert "working_directory" in result.data
def test_null_timeout_like_omit(self, system_state):
"""Explicit JSON null timeout must behave like omit (default 120s)."""
tool = BashTool(system_state)
result = tool.execute({
"command": "echo ok",
"timeout": None,
})
assert result.success
assert "ok" in result.data["output"]
assert result.data["exit_code"] == 0
tests/test_edit_tool.py¶
"""
Test cases for Edit tool
Tests all features from tools.json
"""
import pytest
from pathlib import Path
from tools.edit_tool import EditTool
class TestEditTool:
"""Test Edit tool functionality"""
def test_basic_edit(self, system_state, sample_files):
"""Test basic search and replace"""
tool = EditTool(system_state)
file_path = sample_files["python_file"]
result = tool.execute({
"file_path": str(file_path),
"old_string": "Hello, {name}!",
"new_string": "Hi, {name}!"
})
assert result.success
assert result.data["replacements"] == 1
assert "Hi, {name}!" in file_path.read_text()
def test_replace_all_flag(self, system_state, temp_dir):
"""Test replace_all parameter"""
tool = EditTool(system_state)
file_path = temp_dir / "multi.txt"
file_path.write_text("foo bar foo baz foo")
result = tool.execute({
"file_path": str(file_path),
"old_string": "foo",
"new_string": "replaced",
"replace_all": True
})
assert result.success
assert result.data["replacements"] == 3
assert file_path.read_text() == "replaced bar replaced baz replaced"
def test_uniqueness_check(self, system_state, temp_dir):
"""Test that Edit fails if old_string is not unique (without replace_all)"""
tool = EditTool(system_state)
file_path = temp_dir / "multi.txt"
file_path.write_text("foo bar foo baz foo")
result = tool.execute({
"file_path": str(file_path),
"old_string": "foo",
"new_string": "replaced",
"replace_all": False
})
assert "error" in result.data
assert "appears 3 times" in result.data["error"]
def test_string_not_found(self, system_state, sample_files):
"""Test error when old_string not found"""
tool = EditTool(system_state)
result = tool.execute({
"file_path": str(sample_files["python_file"]),
"old_string": "NONEXISTENT_STRING_12345",
"new_string": "replacement"
})
assert "error" in result.data
assert "not found" in result.data["error"].lower()
def test_preserve_indentation(self, system_state, temp_dir):
"""Test that indentation is preserved"""
tool = EditTool(system_state)
file_path = temp_dir / "indent.py"
file_path.write_text("""
def function():
if True:
print("hello")
""")
result = tool.execute({
"file_path": str(file_path),
"old_string": ' print("hello")',
"new_string": ' print("world")'
})
assert result.success
content = file_path.read_text()
assert ' print("world")' in content # 8 spaces preserved
def test_multiline_replacement(self, system_state, temp_dir):
"""Test replacing multiline strings"""
tool = EditTool(system_state)
file_path = temp_dir / "multi.txt"
file_path.write_text("Line 1\nLine 2\nLine 3\nLine 4")
result = tool.execute({
"file_path": str(file_path),
"old_string": "Line 2\nLine 3",
"new_string": "Replaced Lines"
})
assert result.success
assert "Replaced Lines" in file_path.read_text()
def test_file_not_found(self, system_state):
"""Test error when file doesn't exist"""
tool = EditTool(system_state)
result = tool.execute({
"file_path": "/nonexistent/file.txt",
"old_string": "old",
"new_string": "new"
})
assert "error" in result.data
assert "not found" in result.data["error"].lower()
def test_lint_check_after_edit(self, system_state, temp_dir):
"""Test that lint check runs after Python file edit"""
tool = EditTool(system_state)
file_path = temp_dir / "test.py"
file_path.write_text("def hello():\n return 'world'\n")
result = tool.execute({
"file_path": str(file_path),
"old_string": "return 'world'",
"new_string": "return 'universe'"
})
assert result.success
assert "lint_check" in result.data
assert not result.data["lint_check"]["has_errors"]
def test_edit_creates_syntax_error(self, system_state, temp_dir):
"""Test lint check detects errors introduced by edit"""
tool = EditTool(system_state)
file_path = temp_dir / "test.py"
file_path.write_text("def hello():\n return 'world'\n")
result = tool.execute({
"file_path": str(file_path),
"old_string": "return 'world'",
"new_string": "return 'world" # Missing closing quote
})
assert result.success # Edit succeeds
assert "lint_check" in result.data
assert result.data["lint_check"]["has_errors"]
def test_length_tracking(self, system_state, temp_dir):
"""Test old_length and new_length tracking"""
tool = EditTool(system_state)
file_path = temp_dir / "test.txt"
file_path.write_text("Short text")
result = tool.execute({
"file_path": str(file_path),
"old_string": "Short",
"new_string": "Very long expanded"
})
assert result.success
assert result.data["old_length"] < result.data["new_length"]
tests/test_exit_plan_mode_tool.py¶
"""
Test cases for ExitPlanMode tool
Tests all features from tools.json
"""
import pytest
from tools.exit_plan_mode_tool import ExitPlanModeTool
class TestExitPlanModeTool:
"""Test ExitPlanMode tool functionality"""
def test_basic_plan_submission(self, system_state):
"""Test submitting a plan"""
tool = ExitPlanModeTool(system_state)
plan = """
## Implementation Plan
1. Create database schema
2. Implement API endpoints
3. Write tests
"""
result = tool.execute({
"plan": plan
})
assert result.success
assert result.data["action"] == "exit_plan_mode"
assert result.data["plan"] == plan
assert "message" in result.data
def test_markdown_plan(self, system_state):
"""Test that plan supports markdown"""
tool = ExitPlanModeTool(system_state)
plan = """
# Implementation Plan
## Phase 1
- [ ] Task 1
- [ ] Task 2
## Phase 2
- [ ] Task 3
**Note**: This is a markdown plan
"""
result = tool.execute({
"plan": plan
})
assert result.success
assert "# Implementation Plan" in result.data["plan"]
assert "**Note**" in result.data["plan"]
def test_empty_plan(self, system_state):
"""Test with empty plan"""
tool = ExitPlanModeTool(system_state)
result = tool.execute({
"plan": ""
})
assert result.success
assert result.data["plan"] == ""
tests/test_glob_tool.py¶
"""
Test cases for Glob tool
Tests all features from tools.json
"""
import pytest
from tools.glob_tool import GlobTool
class TestGlobTool:
"""Test Glob tool functionality"""
def test_basic_glob(self, system_state, sample_files):
"""Test basic glob pattern"""
tool = GlobTool(system_state)
result = tool.execute({
"pattern": "*.py",
"path": str(sample_files["temp_dir"])
})
assert result.success
assert result.data["total_matches"] >= 1
assert any("sample.py" in m for m in result.data["matches"])
def test_recursive_glob(self, system_state, sample_files):
"""Test recursive pattern search"""
tool = GlobTool(system_state)
result = tool.execute({
"pattern": "**/*.py",
"path": str(sample_files["temp_dir"])
})
assert result.success
# Should find both sample.py and nested.py
assert result.data["total_matches"] >= 2
def test_auto_recursive_prefix(self, system_state, sample_files):
"""Test that patterns without **/ are auto-prefixed"""
tool = GlobTool(system_state)
result = tool.execute({
"pattern": "*.py", # Should become **/*.py
"path": str(sample_files["temp_dir"])
})
assert result.success
# Should still find nested files
assert result.data["total_matches"] >= 1
def test_sorted_by_modification_time(self, system_state, sample_files):
"""Test that results are sorted by modification time"""
tool = GlobTool(system_state)
result = tool.execute({
"pattern": "*.txt",
"path": str(sample_files["temp_dir"])
})
assert result.success
assert result.data["total_matches"] >= 2
# Results should be in a list
assert isinstance(result.data["matches"], list)
def test_no_matches(self, system_state, sample_files):
"""Test when pattern matches nothing"""
tool = GlobTool(system_state)
result = tool.execute({
"pattern": "*.nonexistent",
"path": str(sample_files["temp_dir"])
})
assert result.success
assert result.data["total_matches"] == 0
assert result.data["matches"] == []
def test_nonexistent_path(self, system_state):
"""Test with nonexistent path"""
tool = GlobTool(system_state)
result = tool.execute({
"pattern": "*.py",
"path": "/nonexistent/path"
})
assert "error" in result.data
def test_not_a_directory(self, system_state, sample_files):
"""Test with a file path instead of directory"""
tool = GlobTool(system_state)
result = tool.execute({
"pattern": "*.py",
"path": str(sample_files["python_file"])
})
assert "error" in result.data
def test_complex_pattern(self, system_state, sample_files):
"""Test complex glob patterns"""
tool = GlobTool(system_state)
result = tool.execute({
"pattern": "**/*.{py,js}",
"path": str(sample_files["temp_dir"])
})
# May or may not work depending on glob implementation
# This tests the behavior
assert result.success or "error" in result.data
def test_default_path(self, system_state, sample_files):
"""Test omitting path parameter uses current directory"""
import os
original_cwd = os.getcwd()
try:
os.chdir(sample_files["temp_dir"])
tool = GlobTool(system_state)
result = tool.execute({
"pattern": "*.py"
})
assert result.success
assert result.data["total_matches"] >= 1
finally:
os.chdir(original_cwd)
def test_null_path_like_omit(self, system_state, sample_files):
"""Explicit JSON null path must behave like omit (cwd / search root)."""
tool = GlobTool(system_state)
result = tool.execute({
"pattern": "*.py",
"path": None,
})
assert result.success
assert "error" not in result.data
assert isinstance(result.data["matches"], list)
tests/test_grep_tool.py¶
"""
Test cases for Grep tool - Pure Python implementation
Tests all features from tools.json
"""
import pytest
from tools.grep_tool import GrepTool
class TestGrepTool:
"""Test Grep tool functionality"""
def test_basic_search(self, system_state, sample_files):
"""Test basic pattern search"""
tool = GrepTool(system_state)
result = tool.execute({
"pattern": "ERROR",
"path": str(sample_files["temp_dir"]),
"output_mode": "files_with_matches"
})
assert result.success
assert "file1.txt" in result.data["output"]
assert result.data["matches"] >= 1
def test_case_insensitive_search(self, system_state, sample_files):
"""Test -i flag for case insensitive search"""
tool = GrepTool(system_state)
result = tool.execute({
"pattern": "error", # lowercase
"path": str(sample_files["temp_dir"]),
"output_mode": "files_with_matches",
"-i": True
})
assert result.success
assert "file1.txt" in result.data["output"]
def test_content_output_mode(self, system_state, sample_files):
"""Test output_mode: content shows matching lines"""
tool = GrepTool(system_state)
result = tool.execute({
"pattern": "ERROR",
"path": str(sample_files["temp_dir"]),
"output_mode": "content"
})
assert result.success
assert "ERROR" in result.data["output"]
assert "file1.txt" in result.data["output"]
def test_content_with_line_numbers(self, system_state, sample_files):
"""Test -n flag for line numbers"""
tool = GrepTool(system_state)
result = tool.execute({
"pattern": "ERROR",
"path": str(sample_files["temp_dir"]),
"output_mode": "content",
"-n": True
})
assert result.success
output = result.data["output"]
# Should contain line numbers in format "3:"
assert ":" in output
def test_context_lines_after(self, system_state, sample_files):
"""Test -A flag for context after match"""
tool = GrepTool(system_state)
result = tool.execute({
"pattern": "ERROR",
"path": str(sample_files["temp_dir"]),
"output_mode": "content",
"-A": 1
})
assert result.success
# Should show line with ERROR and one line after
assert "ERROR" in result.data["output"]
def test_context_lines_before(self, system_state, sample_files):
"""Test -B flag for context before match"""
tool = GrepTool(system_state)
result = tool.execute({
"pattern": "ERROR",
"path": str(sample_files["temp_dir"]),
"output_mode": "content",
"-B": 1
})
assert result.success
assert "ERROR" in result.data["output"]
def test_context_lines_around(self, system_state, sample_files):
"""Test -C flag for context around match"""
tool = GrepTool(system_state)
result = tool.execute({
"pattern": "ERROR",
"path": str(sample_files["temp_dir"]),
"output_mode": "content",
"-C": 2
})
assert result.success
assert "ERROR" in result.data["output"]
def test_count_output_mode(self, system_state, sample_files):
"""Test output_mode: count shows match counts per file"""
tool = GrepTool(system_state)
result = tool.execute({
"pattern": "ERROR",
"path": str(sample_files["temp_dir"]),
"output_mode": "count"
})
assert result.success
# Should show file:count format
assert "file1.txt:1" in result.data["output"]
def test_glob_filtering(self, system_state, sample_files):
"""Test glob parameter to filter files"""
tool = GrepTool(system_state)
result = tool.execute({
"pattern": "def",
"path": str(sample_files["temp_dir"]),
"glob": "*.py",
"output_mode": "files_with_matches"
})
assert result.success
assert ".py" in result.data["output"]
assert ".txt" not in result.data["output"]
def test_type_filtering(self, system_state, sample_files):
"""Test type parameter for file type filtering"""
tool = GrepTool(system_state)
result = tool.execute({
"pattern": "def",
"path": str(sample_files["temp_dir"]),
"type": "py",
"output_mode": "files_with_matches"
})
assert result.success
assert "sample.py" in result.data["output"]
def test_head_limit(self, system_state, sample_files):
"""Test head_limit parameter"""
tool = GrepTool(system_state)
result = tool.execute({
"pattern": ".", # Match everything
"path": str(sample_files["temp_dir"]),
"output_mode": "files_with_matches",
"head_limit": 1
})
assert result.success
# Should only return 1 file
files = result.data["output"].strip().split('\n')
assert len(files) <= 1
def test_regex_pattern(self, system_state, sample_files):
"""Test full regex syntax support"""
tool = GrepTool(system_state)
result = tool.execute({
"pattern": r"def\s+\w+", # Match function definitions
"path": str(sample_files["temp_dir"]),
"output_mode": "content"
})
assert result.success
assert "def" in result.data["output"]
def test_multiline_mode(self, system_state, sample_files):
"""Test multiline mode"""
tool = GrepTool(system_state)
# Create a file with multiline pattern
multiline_file = sample_files["temp_dir"] / "multiline.txt"
multiline_file.write_text("Start\nMiddle\nEnd")
result = tool.execute({
"pattern": r"Start.*End",
"path": str(sample_files["temp_dir"]),
"multiline": True,
"output_mode": "content"
})
assert result.success
def test_no_matches(self, system_state, sample_files):
"""Test when pattern matches nothing"""
tool = GrepTool(system_state)
result = tool.execute({
"pattern": "NONEXISTENT_PATTERN_12345",
"path": str(sample_files["temp_dir"]),
"output_mode": "files_with_matches"
})
assert result.success
assert "No matches found" in result.data["output"]
assert result.data["matches"] == 0
def test_invalid_regex(self, system_state, sample_files):
"""Test invalid regex pattern"""
tool = GrepTool(system_state)
result = tool.execute({
"pattern": "[invalid(", # Invalid regex
"path": str(sample_files["temp_dir"])
})
# Tool-level errors are reported in data (success stays True unless
# _execute_impl raises) — same convention as all other error tests.
assert "error" in result.data
assert "invalid regex" in result.data["error"].lower()
def test_nonexistent_path(self, system_state):
"""Test searching in nonexistent path"""
tool = GrepTool(system_state)
result = tool.execute({
"pattern": "test",
"path": "/nonexistent/path/12345"
})
assert "error" in result.data
def test_single_file_search(self, system_state, sample_files):
"""Test searching a single file"""
tool = GrepTool(system_state)
result = tool.execute({
"pattern": "ERROR",
"path": str(sample_files["text_file1"]),
"output_mode": "content"
})
assert result.success
assert "ERROR" in result.data["output"]
def test_null_context_before_like_omit(self, system_state, sample_files):
"""Explicit JSON null -B must behave like omit (default 0)."""
tool = GrepTool(system_state)
result = tool.execute({
"pattern": "ERROR",
"path": str(sample_files["text_file1"]),
"output_mode": "content",
"-B": None,
})
assert result.success
assert "ERROR" in result.data["output"]
def test_null_context_after_like_omit(self, system_state, sample_files):
"""Explicit JSON null -A must behave like omit (default 0)."""
tool = GrepTool(system_state)
result = tool.execute({
"pattern": "ERROR",
"path": str(sample_files["text_file1"]),
"output_mode": "content",
"-A": None,
})
assert result.success
assert "ERROR" in result.data["output"]
def test_null_context_around_like_omit(self, system_state, sample_files):
"""Explicit JSON null -C must behave like omit (default 0)."""
tool = GrepTool(system_state)
result = tool.execute({
"pattern": "ERROR",
"path": str(sample_files["text_file1"]),
"output_mode": "content",
"-C": None,
})
assert result.success
assert "ERROR" in result.data["output"]
def test_null_path_like_omit(self, system_state, sample_files):
"""Explicit JSON null path must behave like omit (default search root)."""
tool = GrepTool(system_state)
result = tool.execute({
"pattern": "ERROR",
"path": None,
"output_mode": "content",
})
assert result.success
assert "error" not in result.data
tests/test_integration.py¶
"""
Integration tests for the complete agent system
Tests system hints, tool chaining, and end-to-end workflows
"""
import pytest
from system_state import SystemState
from tools.grep_tool import GrepTool
from tools.write_tool import WriteTool
from tools.todo_write_tool import TodoWriteTool
class TestSystemHints:
"""Test system hint generation"""
def test_system_hint_structure(self, system_state):
"""Test that system hint includes all required sections"""
hint = system_state.get_system_hint()
assert "# System State" in hint
assert "Current Time:" in hint
assert "Working Directory:" in hint
assert "OS:" in hint
assert "Python:" in hint
def test_tool_call_statistics_in_hint(self, system_state):
"""Test that tool calls are tracked in system hint"""
# Make some tool calls
grep_tool = GrepTool(system_state)
grep_tool.execute({"pattern": "test", "path": "."})
grep_tool.execute({"pattern": "test2", "path": "."})
hint = system_state.get_system_hint()
assert "# Tool Call Statistics" in hint
assert "Grep: 2 calls" in hint
def test_tool_warning_after_three_calls(self, system_state):
"""Test that system hint warns after 3+ tool calls"""
tool = GrepTool(system_state)
# Call tool 4 times
for i in range(4):
tool.execute({"pattern": f"test{i}", "path": "."})
hint = system_state.get_system_hint()
assert "⚠️" in hint
assert "4 times" in hint
assert "Consider alternative approaches" in hint
def test_todo_list_in_hint(self, system_state):
"""Test that TODO list appears in system hint"""
todo_tool = TodoWriteTool(system_state)
todos = [
{"id": "1", "content": "Task 1", "status": "completed"},
{"id": "2", "content": "Task 2", "status": "in_progress"},
{"id": "3", "content": "Task 3", "status": "pending"}
]
todo_tool.execute({"todos": todos})
hint = system_state.get_system_hint()
assert "# Current TODO List" in hint
assert "✅" in hint # Completed
assert "🔄" in hint # In progress
assert "⬜" in hint # Pending
assert "Task 1" in hint
assert "Task 2" in hint
assert "Task 3" in hint
class TestToolChaining:
"""Test chaining multiple tools together"""
def test_write_then_read_workflow(self, system_state, temp_dir):
"""Test writing a file then reading it back"""
from tools.write_tool import WriteTool
from tools.read_tool import ReadTool
write_tool = WriteTool(system_state)
read_tool = ReadTool(system_state)
file_path = temp_dir / "chained.txt"
content = "This is a test"
# Write file
write_result = write_tool.execute({
"file_path": str(file_path),
"content": content
})
assert write_result.success
# Read it back
read_result = read_tool.execute({
"file_path": str(file_path)
})
assert read_result.success
assert content in read_result.data["content"]
def test_write_search_edit_workflow(self, system_state, temp_dir):
"""Test complete workflow: write, search, edit"""
from tools.write_tool import WriteTool
from tools.grep_tool import GrepTool
from tools.edit_tool import EditTool
write_tool = WriteTool(system_state)
grep_tool = GrepTool(system_state)
edit_tool = EditTool(system_state)
file_path = temp_dir / "workflow.py"
# 1. Write initial file
write_result = write_tool.execute({
"file_path": str(file_path),
"content": "def old_function():\n return 'old'\n"
})
assert write_result.success
# 2. Search for pattern
grep_result = grep_tool.execute({
"pattern": "old_function",
"path": str(temp_dir),
"output_mode": "files_with_matches"
})
assert grep_result.success
assert str(file_path) in grep_result.data["output"]
# 3. Edit the file
edit_result = edit_tool.execute({
"file_path": str(file_path),
"old_string": "old_function",
"new_string": "new_function"
})
assert edit_result.success
# 4. Verify change with another search
grep_result2 = grep_tool.execute({
"pattern": "new_function",
"path": str(temp_dir),
"output_mode": "content"
})
assert grep_result2.success
assert "new_function" in grep_result2.data["output"]
def test_metadata_consistency(self, system_state):
"""Test that metadata is consistent across tool calls"""
tool = GrepTool(system_state)
# First call
result1 = tool.execute({"pattern": "test", "path": "."})
assert result1.metadata["call_number"] == 1
assert result1.metadata["tool"] == "Grep"
# Second call
result2 = tool.execute({"pattern": "test2", "path": "."})
assert result2.metadata["call_number"] == 2
# Third call
result3 = tool.execute({"pattern": "test3", "path": "."})
assert result3.metadata["call_number"] == 3
tests/test_kill_bash_tool.py¶
"""
Test cases for KillBash tool
Tests all features from tools.json
"""
import pytest
from tools.kill_bash_tool import KillBashTool
from tools.bash_tool import BashTool
class TestKillBashTool:
"""Test KillBash tool functionality"""
def test_kill_shell_session(self, system_state):
"""Test killing a shell session"""
bash_tool = BashTool(system_state)
kill_tool = KillBashTool(system_state)
# Create a shell session
bash_tool.execute({"command": "echo test"})
shell_id = "default"
# Verify session exists
assert shell_id in system_state.shell_sessions
# Kill the session
result = kill_tool.execute({
"shell_id": shell_id
})
assert result.success
assert result.data["status"] == "terminated"
assert shell_id not in system_state.shell_sessions
def test_kill_nonexistent_session(self, system_state):
"""Test error when trying to kill nonexistent session"""
tool = KillBashTool(system_state)
result = tool.execute({
"shell_id": "nonexistent_session"
})
assert "error" in result.data
assert "not found" in result.data["error"]
def test_shell_id_returned(self, system_state):
"""Test that shell_id is included in response"""
bash_tool = BashTool(system_state)
kill_tool = KillBashTool(system_state)
bash_tool.execute({"command": "echo test"})
result = kill_tool.execute({
"shell_id": "default"
})
assert result.success
assert result.data["shell_id"] == "default"
tests/test_ls_tool.py¶
"""
Test cases for LS tool
Tests all features from tools.json
"""
import pytest
from pathlib import Path
from tools.ls_tool import LSTool
class TestLSTool:
"""Test LS tool functionality"""
def test_basic_listing(self, system_state, sample_files):
"""Test basic directory listing"""
tool = LSTool(system_state)
result = tool.execute({
"path": str(sample_files["temp_dir"])
})
assert result.success
assert result.data["total_entries"] >= 2
# Check entries structure
entries = result.data["entries"]
assert all("name" in e for e in entries)
assert all("type" in e for e in entries)
assert all("size" in e for e in entries)
assert all("path" in e for e in entries)
def test_files_and_directories(self, system_state, sample_files):
"""Test that both files and directories are listed"""
tool = LSTool(system_state)
result = tool.execute({
"path": str(sample_files["temp_dir"])
})
assert result.success
entries = result.data["entries"]
# Should have files
files = [e for e in entries if e["type"] == "file"]
assert len(files) > 0
# Should have directories
dirs = [e for e in entries if e["type"] == "dir"]
assert len(dirs) > 0
def test_hidden_files_excluded(self, system_state, temp_dir):
"""Test that hidden files (starting with .) are excluded"""
tool = LSTool(system_state)
# Create hidden file
hidden_file = temp_dir / ".hidden"
hidden_file.write_text("secret")
# Create normal file
normal_file = temp_dir / "normal.txt"
normal_file.write_text("public")
result = tool.execute({
"path": str(temp_dir)
})
assert result.success
entry_names = [e["name"] for e in result.data["entries"]]
assert "normal.txt" in entry_names
assert ".hidden" not in entry_names
def test_ignore_patterns(self, system_state, temp_dir):
"""Test ignore parameter with glob patterns"""
tool = LSTool(system_state)
# Create various files
(temp_dir / "keep.txt").write_text("keep")
(temp_dir / "ignore.log").write_text("ignore")
(temp_dir / "also_keep.py").write_text("keep")
result = tool.execute({
"path": str(temp_dir),
"ignore": ["*.log"]
})
assert result.success
entry_names = [e["name"] for e in result.data["entries"]]
assert "keep.txt" in entry_names
assert "also_keep.py" in entry_names
assert "ignore.log" not in entry_names
def test_multiple_ignore_patterns(self, system_state, temp_dir):
"""Test multiple ignore patterns"""
tool = LSTool(system_state)
(temp_dir / "file.txt").write_text("1")
(temp_dir / "file.log").write_text("2")
(temp_dir / "file.tmp").write_text("3")
result = tool.execute({
"path": str(temp_dir),
"ignore": ["*.log", "*.tmp"]
})
assert result.success
entry_names = [e["name"] for e in result.data["entries"]]
assert "file.txt" in entry_names
assert "file.log" not in entry_names
assert "file.tmp" not in entry_names
def test_sorted_output(self, system_state, temp_dir):
"""Test that entries are sorted"""
tool = LSTool(system_state)
# Create files in specific order
(temp_dir / "z_file.txt").write_text("1")
(temp_dir / "a_file.txt").write_text("2")
(temp_dir / "m_file.txt").write_text("3")
result = tool.execute({
"path": str(temp_dir)
})
assert result.success
entry_names = [e["name"] for e in result.data["entries"]]
# Should be sorted alphabetically
sorted_names = sorted(entry_names)
assert entry_names == sorted_names
def test_file_sizes(self, system_state, temp_dir):
"""Test that file sizes are reported"""
tool = LSTool(system_state)
file_path = temp_dir / "sized.txt"
content = "A" * 1000
file_path.write_text(content)
result = tool.execute({
"path": str(temp_dir)
})
assert result.success
entry = next(e for e in result.data["entries"] if e["name"] == "sized.txt")
assert entry["size"] == 1000
def test_directory_size_zero(self, system_state, temp_dir):
"""Test that directories have size 0"""
tool = LSTool(system_state)
subdir = temp_dir / "subdir"
subdir.mkdir()
result = tool.execute({
"path": str(temp_dir)
})
assert result.success
dir_entry = next(e for e in result.data["entries"] if e["name"] == "subdir")
assert dir_entry["type"] == "dir"
assert dir_entry["size"] == 0
def test_path_not_found(self, system_state):
"""Test error when path doesn't exist"""
tool = LSTool(system_state)
result = tool.execute({
"path": "/nonexistent/path"
})
assert "error" in result.data
assert "not found" in result.data["error"].lower()
def test_not_a_directory(self, system_state, sample_files):
"""Test error when path is a file not directory"""
tool = LSTool(system_state)
result = tool.execute({
"path": str(sample_files["python_file"])
})
assert "error" in result.data
assert "not a directory" in result.data["error"].lower()
def test_permission_denied(self, system_state, temp_dir):
"""Test handling of permission errors"""
# This test might not work on all systems
tool = LSTool(system_state)
restricted_dir = temp_dir / "restricted"
restricted_dir.mkdir(mode=0o000)
try:
result = tool.execute({
"path": str(restricted_dir)
})
# Should either succeed (if running as root) or fail with permission error
if "error" in result.data:
assert "permission" in result.data["error"].lower()
finally:
restricted_dir.chmod(0o755) # Restore permissions for cleanup
def test_ignore_null_lists_directory(self, system_state, temp_dir):
"""JSON null ignore must not break listing (agent omits optional array)."""
tool = LSTool(system_state)
(temp_dir / "keep.txt").write_text("keep")
result = tool.execute({
"path": str(temp_dir),
"ignore": None,
})
assert result.success
assert "error" not in result.data
assert any(e["name"] == "keep.txt" for e in result.data["entries"])
tests/test_multi_edit_tool.py¶
"""
Test cases for MultiEdit tool
Tests all features from tools.json
"""
import pytest
from pathlib import Path
from tools.multi_edit_tool import MultiEditTool
class TestMultiEditTool:
"""Test MultiEdit tool functionality"""
def test_multiple_edits(self, system_state, temp_dir):
"""Test multiple edits in one operation"""
tool = MultiEditTool(system_state)
file_path = temp_dir / "multi.py"
file_path.write_text("""
def old_function():
old_var = 1
return old_var
""")
result = tool.execute({
"file_path": str(file_path),
"edits": [
{
"old_string": "old_function",
"new_string": "new_function"
},
{
"old_string": "old_var",
"new_string": "new_var",
"replace_all": True
}
]
})
assert result.success
assert result.data["total_edits"] == 2
assert result.data["successful_edits"] == 2
content = file_path.read_text()
assert "new_function" in content
assert "new_var" in content
assert "old_var" not in content
def test_sequential_application(self, system_state, temp_dir):
"""Test that edits are applied sequentially"""
tool = MultiEditTool(system_state)
file_path = temp_dir / "seq.txt"
file_path.write_text("A B C")
result = tool.execute({
"file_path": str(file_path),
"edits": [
{"old_string": "A", "new_string": "X"},
{"old_string": "X B", "new_string": "Y"}, # Depends on first edit
{"old_string": "Y C", "new_string": "Z"} # Depends on second edit
]
})
assert result.success
assert file_path.read_text() == "Z"
def test_atomic_edits(self, system_state, temp_dir):
"""Test that if any edit fails, none are applied"""
tool = MultiEditTool(system_state)
file_path = temp_dir / "atomic.txt"
original = "First line\nSecond line\n"
file_path.write_text(original)
result = tool.execute({
"file_path": str(file_path),
"edits": [
{"old_string": "First", "new_string": "1st"},
{"old_string": "NONEXISTENT", "new_string": "X"}, # This will fail
{"old_string": "Second", "new_string": "2nd"}
]
})
# Should fail
assert "error" in result.data
assert result.data["completed_edits"] == 1
# File should be modified (edits are not rolled back in current implementation)
def test_file_creation(self, system_state, temp_dir):
"""Test creating new file with MultiEdit (empty old_string in first edit)"""
tool = MultiEditTool(system_state)
file_path = temp_dir / "new_file.py"
result = tool.execute({
"file_path": str(file_path),
"edits": [
{
"old_string": "",
"new_string": "def hello():\n pass\n"
}
]
})
assert result.success
assert file_path.exists()
assert "def hello" in file_path.read_text()
assert result.data["edit_results"][0]["action"] == "created"
def test_create_and_modify(self, system_state, temp_dir):
"""Test creating file and then modifying it in subsequent edits"""
tool = MultiEditTool(system_state)
file_path = temp_dir / "new_file.py"
result = tool.execute({
"file_path": str(file_path),
"edits": [
{
"old_string": "",
"new_string": "def old_name():\n pass\n"
},
{
"old_string": "old_name",
"new_string": "new_name"
}
]
})
assert result.success
assert "new_name" in file_path.read_text()
assert "old_name" not in file_path.read_text()
def test_replace_all_in_multi_edit(self, system_state, temp_dir):
"""Test replace_all in one of multiple edits"""
tool = MultiEditTool(system_state)
file_path = temp_dir / "test.txt"
file_path.write_text("foo bar foo baz foo")
result = tool.execute({
"file_path": str(file_path),
"edits": [
{
"old_string": "foo",
"new_string": "FOO",
"replace_all": True
},
{
"old_string": "bar",
"new_string": "BAR"
}
]
})
assert result.success
assert file_path.read_text() == "FOO BAR FOO baz FOO"
def test_edit_results_tracking(self, system_state, temp_dir):
"""Test that edit_results tracks each edit"""
tool = MultiEditTool(system_state)
file_path = temp_dir / "track.txt"
file_path.write_text("A B C")
result = tool.execute({
"file_path": str(file_path),
"edits": [
{"old_string": "A", "new_string": "1"},
{"old_string": "B", "new_string": "2"},
{"old_string": "C", "new_string": "3"}
]
})
assert result.success
assert len(result.data["edit_results"]) == 3
assert all(r["success"] for r in result.data["edit_results"])
def test_lint_check_after_multi_edit(self, system_state, temp_dir):
"""Test lint checking after multiple edits"""
tool = MultiEditTool(system_state)
file_path = temp_dir / "test.py"
file_path.write_text("x = 1\ny = 2\n")
result = tool.execute({
"file_path": str(file_path),
"edits": [
{"old_string": "x = 1", "new_string": "x = 10"},
{"old_string": "y = 2", "new_string": "y = 20"}
]
})
assert result.success
assert "lint_check" in result.data
assert not result.data["lint_check"]["has_errors"]
def test_size_tracking(self, system_state, temp_dir):
"""Test old_size and new_size tracking"""
tool = MultiEditTool(system_state)
file_path = temp_dir / "size.txt"
file_path.write_text("Short")
result = tool.execute({
"file_path": str(file_path),
"edits": [
{"old_string": "Short", "new_string": "Very long text here"}
]
})
assert result.success
assert result.data["old_size"] < result.data["new_size"]
def test_null_edits_like_empty(self, system_state, temp_dir):
"""Explicit JSON null edits must behave like an empty list."""
tool = MultiEditTool(system_state)
file_path = temp_dir / "null_edits.py"
file_path.write_text("x = 1\n")
result = tool.execute({
"file_path": str(file_path),
"edits": None,
})
assert result.success
assert "error" not in result.data
assert result.data["total_edits"] == 0
assert file_path.read_text() == "x = 1\n"
tests/test_notebook_edit_tool.py¶
"""
Test cases for NotebookEdit tool
Tests all features from tools.json
"""
import pytest
import json
from pathlib import Path
from tools.notebook_edit_tool import NotebookEditTool
@pytest.fixture
def sample_notebook(temp_dir):
"""Create a sample Jupyter notebook"""
notebook_path = temp_dir / "test.ipynb"
notebook_data = {
"cells": [
{
"id": "cell-1",
"cell_type": "code",
"source": ["print('hello')"],
"outputs": [],
"execution_count": None
},
{
"id": "cell-2",
"cell_type": "markdown",
"source": ["# Title"]
},
{
"id": "cell-3",
"cell_type": "code",
"source": ["x = 1\n", "y = 2"],
"outputs": [],
"execution_count": None
}
],
"metadata": {},
"nbformat": 4,
"nbformat_minor": 2
}
notebook_path.write_text(json.dumps(notebook_data, indent=2))
return notebook_path
class TestNotebookEditTool:
"""Test NotebookEdit tool functionality"""
def test_replace_cell(self, system_state, sample_notebook):
"""Test edit_mode=replace (default)"""
tool = NotebookEditTool(system_state)
result = tool.execute({
"notebook_path": str(sample_notebook),
"cell_id": "cell-1",
"new_source": "print('world')",
"edit_mode": "replace"
})
assert result.success
assert result.data["action"] == "replaced"
# Verify change
notebook = json.loads(sample_notebook.read_text())
cell = next(c for c in notebook["cells"] if c.get("id") == "cell-1")
assert "world" in ''.join(cell["source"])
def test_insert_cell(self, system_state, sample_notebook):
"""Test edit_mode=insert"""
tool = NotebookEditTool(system_state)
result = tool.execute({
"notebook_path": str(sample_notebook),
"cell_id": "cell-1",
"new_source": "# New cell",
"cell_type": "markdown",
"edit_mode": "insert"
})
assert result.success
assert result.data["action"] == "inserted"
# Verify insertion
notebook = json.loads(sample_notebook.read_text())
# Should have 4 cells now (3 original + 1 inserted)
assert len(notebook["cells"]) == 4
def test_delete_cell(self, system_state, sample_notebook):
"""Test edit_mode=delete"""
tool = NotebookEditTool(system_state)
result = tool.execute({
"notebook_path": str(sample_notebook),
"cell_id": "cell-2",
"new_source": "", # Not used for delete
"edit_mode": "delete"
})
assert result.success
assert result.data["action"] == "deleted"
# Verify deletion
notebook = json.loads(sample_notebook.read_text())
assert len(notebook["cells"]) == 2
assert not any(c.get("id") == "cell-2" for c in notebook["cells"])
def test_insert_at_beginning(self, system_state, sample_notebook):
"""Test inserting at beginning when cell_id not specified"""
tool = NotebookEditTool(system_state)
result = tool.execute({
"notebook_path": str(sample_notebook),
"new_source": "# First cell",
"cell_type": "markdown",
"edit_mode": "insert"
})
assert result.success
# Verify it was inserted at beginning
notebook = json.loads(sample_notebook.read_text())
assert "First cell" in ''.join(notebook["cells"][0]["source"])
def test_change_cell_type(self, system_state, sample_notebook):
"""Test changing cell type during replace"""
tool = NotebookEditTool(system_state)
result = tool.execute({
"notebook_path": str(sample_notebook),
"cell_id": "cell-1",
"new_source": "# Now markdown",
"cell_type": "markdown",
"edit_mode": "replace"
})
assert result.success
# Verify cell type changed
notebook = json.loads(sample_notebook.read_text())
cell = next(c for c in notebook["cells"] if c.get("id") == "cell-1")
assert cell["cell_type"] == "markdown"
def test_multiline_source(self, system_state, sample_notebook):
"""Test editing with multiline source"""
tool = NotebookEditTool(system_state)
multiline_source = "def hello():\n print('world')\n return True"
result = tool.execute({
"notebook_path": str(sample_notebook),
"cell_id": "cell-1",
"new_source": multiline_source,
"edit_mode": "replace"
})
assert result.success
# Verify multiline source was saved correctly
notebook = json.loads(sample_notebook.read_text())
cell = next(c for c in notebook["cells"] if c.get("id") == "cell-1")
assert len(cell["source"]) == 3
def test_cell_not_found(self, system_state, sample_notebook):
"""Test error when cell_id doesn't exist"""
tool = NotebookEditTool(system_state)
result = tool.execute({
"notebook_path": str(sample_notebook),
"cell_id": "nonexistent-cell",
"new_source": "test",
"edit_mode": "replace"
})
assert "error" in result.data
assert "not found" in result.data["error"]
def test_notebook_not_found(self, system_state):
"""Test error when notebook doesn't exist"""
tool = NotebookEditTool(system_state)
result = tool.execute({
"notebook_path": "/nonexistent/notebook.ipynb",
"cell_id": "cell-1",
"new_source": "test"
})
assert "error" in result.data
assert "not found" in result.data["error"].lower()
def test_invalid_notebook_format(self, system_state, temp_dir):
"""Test error with invalid JSON notebook"""
tool = NotebookEditTool(system_state)
bad_notebook = temp_dir / "bad.ipynb"
bad_notebook.write_text("not valid json")
result = tool.execute({
"notebook_path": str(bad_notebook),
"cell_id": "cell-1",
"new_source": "test"
})
assert "error" in result.data
assert "Invalid Jupyter notebook" in result.data["error"]
def test_delete_requires_cell_id(self, system_state, sample_notebook):
"""Test that delete mode requires cell_id"""
tool = NotebookEditTool(system_state)
result = tool.execute({
"notebook_path": str(sample_notebook),
"new_source": "",
"edit_mode": "delete"
})
assert "error" in result.data
assert "cell_id required" in result.data["error"]
def test_replace_requires_cell_id(self, system_state, sample_notebook):
"""Test that replace mode requires cell_id"""
tool = NotebookEditTool(system_state)
result = tool.execute({
"notebook_path": str(sample_notebook),
"new_source": "test",
"edit_mode": "replace"
})
assert "error" in result.data
assert "cell_id required" in result.data["error"]
def test_delete_without_new_source(self, system_state, sample_notebook):
"""Delete must work when new_source is omitted."""
tool = NotebookEditTool(system_state)
result = tool.execute({
"notebook_path": str(sample_notebook),
"cell_id": "cell-1",
"edit_mode": "delete",
})
assert result.success
assert result.data["action"] == "deleted"
tests/test_read_tool.py¶
"""
Test cases for Read tool
Tests all features from tools.json including images, PDFs, notebooks
"""
import pytest
import json
from pathlib import Path
from tools.read_tool import ReadTool
class TestReadTool:
"""Test Read tool functionality"""
def test_basic_read(self, system_state, sample_files):
"""Test basic file reading"""
tool = ReadTool(system_state)
result = tool.execute({
"file_path": str(sample_files["python_file"])
})
assert result.success
assert "def hello" in result.data["content"]
assert "total_lines" in result.data
def test_line_numbers_format(self, system_state, sample_files):
"""Test cat -n format with line numbers starting at 1"""
tool = ReadTool(system_state)
result = tool.execute({
"file_path": str(sample_files["python_file"])
})
assert result.success
content = result.data["content"]
# Should have format: " 1|line content"
lines = content.split('\n')
first_line = lines[0]
assert "|" in first_line
# Extract line number
line_num = first_line.split('|')[0].strip()
assert line_num.isdigit()
assert int(line_num) >= 1
def test_offset_and_limit(self, system_state, sample_files):
"""Test offset and limit parameters for large files"""
tool = ReadTool(system_state)
# Create a file with many lines
large_file = sample_files["temp_dir"] / "large.txt"
large_file.write_text('\n'.join([f"Line {i}" for i in range(100)]))
result = tool.execute({
"file_path": str(large_file),
"offset": 10,
"limit": 5
})
assert result.success
assert "showing_lines" in result.data
assert "11-15" in result.data["showing_lines"]
# Should have exactly 5 lines
lines = result.data["content"].split('\n')
assert len(lines) == 5
def test_long_line_truncation(self, system_state, sample_files):
"""Test that lines longer than 2000 chars are truncated"""
tool = ReadTool(system_state)
# Create file with very long line
long_file = sample_files["temp_dir"] / "long.txt"
long_line = "A" * 3000
long_file.write_text(long_line)
result = tool.execute({
"file_path": str(long_file)
})
assert result.success
assert "truncated" in result.data["content"]
def test_empty_file(self, system_state, sample_files):
"""Test reading empty file"""
tool = ReadTool(system_state)
empty_file = sample_files["temp_dir"] / "empty.txt"
empty_file.write_text("")
result = tool.execute({
"file_path": str(empty_file)
})
assert result.success
assert "File is empty" in result.data["content"]
def test_nonexistent_file(self, system_state):
"""Test reading nonexistent file"""
tool = ReadTool(system_state)
result = tool.execute({
"file_path": "/nonexistent/file.txt"
})
assert "error" in result.data
assert "not found" in result.data["error"].lower()
def test_binary_file_detection(self, system_state, sample_files):
"""Test binary file detection"""
tool = ReadTool(system_state)
# Create a binary file
binary_file = sample_files["temp_dir"] / "binary.bin"
binary_file.write_bytes(b'\x00\x01\x02\x03\x04\x05')
result = tool.execute({
"file_path": str(binary_file)
})
# Should detect as binary
assert "binary" in result.data.get("error", "").lower()
def test_image_file_handling(self, system_state, sample_files):
"""Test image file handling"""
tool = ReadTool(system_state)
# Create a dummy image file
image_file = sample_files["temp_dir"] / "test.png"
image_file.write_bytes(b'\x89PNG\r\n\x1a\n') # PNG header
result = tool.execute({
"file_path": str(image_file)
})
assert result.success
assert result.data["file_type"] == "image"
assert "PNG" in result.data["format"]
def test_pdf_file_handling(self, system_state, sample_files):
"""Test PDF file handling"""
tool = ReadTool(system_state)
# Create a dummy PDF file
pdf_file = sample_files["temp_dir"] / "test.pdf"
pdf_file.write_bytes(b'%PDF-1.4')
result = tool.execute({
"file_path": str(pdf_file)
})
assert result.success
assert result.data["file_type"] == "pdf"
def test_jupyter_notebook_reading(self, system_state, sample_files):
"""Test Jupyter notebook reading"""
tool = ReadTool(system_state)
# Create a simple notebook
notebook_file = sample_files["temp_dir"] / "test.ipynb"
notebook_data = {
"cells": [
{
"cell_type": "code",
"source": ["print('hello')"],
"outputs": []
},
{
"cell_type": "markdown",
"source": ["# Title"]
}
],
"metadata": {},
"nbformat": 4,
"nbformat_minor": 2
}
notebook_file.write_text(json.dumps(notebook_data))
result = tool.execute({
"file_path": str(notebook_file)
})
assert result.success
assert result.data["file_type"] == "jupyter_notebook"
assert result.data["total_cells"] == 2
assert "hello" in result.data["content"]
def test_not_a_file_error(self, system_state, sample_files):
"""Test reading a directory instead of file"""
tool = ReadTool(system_state)
result = tool.execute({
"file_path": str(sample_files["temp_dir"])
})
assert "error" in result.data
assert "not a file" in result.data["error"].lower()
def test_null_offset_and_limit(self, system_state, sample_files):
"""JSON null offset/limit must use defaults (agent omits optional numbers)."""
tool = ReadTool(system_state)
path = sample_files["python_file"]
result = tool.execute({
"file_path": str(path),
"offset": None,
"limit": None,
})
assert result.success
assert "error" not in result.data
assert "def hello" in result.data["content"]
assert result.data["total_lines"] > 0
tests/test_shell_session.py¶
"""
Test cases for ShellSession __CMD_DONE__ marker handling
"""
from tools.shell_session import ShellSession
class TestShellSessionMarker:
"""Test that command output containing the marker string can't break the protocol"""
def test_basic_command_and_exit_code(self):
"""Normal commands still work and report exit codes"""
s = ShellSession(session_id="test_basic")
try:
out, code = s.execute("echo hello", timeout=10)
assert code == 0
assert "hello" in out
out, code = s.execute("false", timeout=10)
assert code == 1
finally:
s.kill()
def test_output_containing_marker_text(self):
"""Output containing the marker text must not crash or be truncated"""
s = ShellSession(session_id="test_marker_text")
try:
out, code = s.execute('echo "prefix__CMD_DONE__notanumber"', timeout=10)
assert code == 0
assert "prefix__CMD_DONE__notanumber" in out
finally:
s.kill()
def test_marker_text_does_not_desync_next_command(self):
"""Marker-like output must not swallow the next command's output"""
s = ShellSession(session_id="test_desync")
try:
s.execute('echo "see __CMD_DONE__123 here"', timeout=10)
out, code = s.execute("echo hello-after", timeout=10)
assert code == 0
assert "hello-after" in out
finally:
s.kill()
tests/test_todo_write_tool.py¶
"""
Test cases for TodoWrite tool
Tests all features from tools.json
"""
import pytest
from tools.todo_write_tool import TodoWriteTool
class TestTodoWriteTool:
"""Test TodoWrite tool functionality"""
def test_create_todo_list(self, system_state):
"""Test creating a TODO list"""
tool = TodoWriteTool(system_state)
todos = [
{"id": "1", "content": "First task", "status": "pending"},
{"id": "2", "content": "Second task", "status": "in_progress"},
{"id": "3", "content": "Third task", "status": "completed"}
]
result = tool.execute({"todos": todos})
assert result.success
assert result.data["total_todos"] == 3
assert result.data["pending"] == 1
assert result.data["in_progress"] == 1
assert result.data["completed"] == 1
# Verify state was updated
assert system_state.todos == todos
def test_update_todo_list(self, system_state):
"""Test updating an existing TODO list"""
tool = TodoWriteTool(system_state)
# Create initial list
initial_todos = [
{"id": "1", "content": "Task 1", "status": "pending"}
]
tool.execute({"todos": initial_todos})
# Update list
updated_todos = [
{"id": "1", "content": "Task 1", "status": "completed"}
]
result = tool.execute({"todos": updated_todos})
assert result.success
assert result.data["completed"] == 1
assert result.data["pending"] == 0
def test_todo_validation_missing_fields(self, system_state):
"""Test validation rejects TODOs missing required fields"""
tool = TodoWriteTool(system_state)
# Missing 'status' field
result = tool.execute({
"todos": [{"id": "1", "content": "Task"}]
})
assert "error" in result.data
assert "must have" in result.data["error"]
def test_todo_validation_invalid_status(self, system_state):
"""Test validation rejects invalid status values"""
tool = TodoWriteTool(system_state)
result = tool.execute({
"todos": [
{"id": "1", "content": "Task", "status": "invalid_status"}
]
})
assert "error" in result.data
assert "Invalid status" in result.data["error"]
def test_valid_status_values(self, system_state):
"""Test all valid status values"""
tool = TodoWriteTool(system_state)
todos = [
{"id": "1", "content": "Task 1", "status": "pending"},
{"id": "2", "content": "Task 2", "status": "in_progress"},
{"id": "3", "content": "Task 3", "status": "completed"}
]
result = tool.execute({"todos": todos})
assert result.success
assert result.data["pending"] == 1
assert result.data["in_progress"] == 1
assert result.data["completed"] == 1
def test_empty_todo_list(self, system_state):
"""Test creating empty TODO list"""
tool = TodoWriteTool(system_state)
result = tool.execute({"todos": []})
assert result.success
assert result.data["total_todos"] == 0
assert result.data["pending"] == 0
assert result.data["in_progress"] == 0
assert result.data["completed"] == 0
def test_statistics_calculation(self, system_state):
"""Test that statistics are calculated correctly"""
tool = TodoWriteTool(system_state)
todos = [
{"id": "1", "content": "A", "status": "pending"},
{"id": "2", "content": "B", "status": "pending"},
{"id": "3", "content": "C", "status": "in_progress"},
{"id": "4", "content": "D", "status": "completed"},
{"id": "5", "content": "E", "status": "completed"},
{"id": "6", "content": "F", "status": "completed"}
]
result = tool.execute({"todos": todos})
assert result.success
assert result.data["total_todos"] == 6
assert result.data["pending"] == 2
assert result.data["in_progress"] == 1
assert result.data["completed"] == 3
def test_null_todos_like_empty(self, system_state):
"""Explicit JSON null todos must behave like an empty list."""
tool = TodoWriteTool(system_state)
result = tool.execute({"todos": None})
assert result.success
assert "error" not in result.data
assert result.data["total_todos"] == 0
tests/test_write_tool.py¶
"""
Test cases for Write tool
Tests all features from tools.json
"""
import pytest
from pathlib import Path
from tools.write_tool import WriteTool
class TestWriteTool:
"""Test Write tool functionality"""
def test_basic_write(self, system_state, temp_dir):
"""Test basic file writing"""
tool = WriteTool(system_state)
file_path = temp_dir / "new_file.txt"
result = tool.execute({
"file_path": str(file_path),
"content": "Hello, World!"
})
assert result.success
assert file_path.exists()
assert file_path.read_text() == "Hello, World!"
assert result.data["bytes_written"] > 0
assert result.data["lines_written"] == 1
def test_overwrite_existing_file(self, system_state, sample_files):
"""Test that Write overwrites existing files"""
tool = WriteTool(system_state)
file_path = sample_files["text_file1"]
original_content = file_path.read_text()
new_content = "New content"
result = tool.execute({
"file_path": str(file_path),
"content": new_content
})
assert result.success
assert file_path.read_text() == new_content
assert file_path.read_text() != original_content
def test_create_parent_directories(self, system_state, temp_dir):
"""Test that Write creates parent directories if needed"""
tool = WriteTool(system_state)
file_path = temp_dir / "deep" / "nested" / "dir" / "file.txt"
result = tool.execute({
"file_path": str(file_path),
"content": "Content"
})
assert result.success
assert file_path.exists()
assert file_path.parent.exists()
def test_multiline_content(self, system_state, temp_dir):
"""Test writing multiline content"""
tool = WriteTool(system_state)
file_path = temp_dir / "multiline.txt"
content = "Line 1\nLine 2\nLine 3\n"
result = tool.execute({
"file_path": str(file_path),
"content": content
})
assert result.success
assert result.data["lines_written"] == 4 # 3 lines + final newline
assert file_path.read_text() == content
def test_python_lint_check_success(self, system_state, temp_dir):
"""Test automatic lint checking for valid Python file"""
tool = WriteTool(system_state)
file_path = temp_dir / "valid.py"
result = tool.execute({
"file_path": str(file_path),
"content": "def hello():\n return 'world'\n"
})
assert result.success
assert "lint_check" in result.data
assert result.data["lint_check"]["language"] == "python"
assert not result.data["lint_check"]["has_errors"]
def test_python_lint_check_failure(self, system_state, temp_dir):
"""Test automatic lint checking for invalid Python file"""
tool = WriteTool(system_state)
file_path = temp_dir / "invalid.py"
result = tool.execute({
"file_path": str(file_path),
"content": "def hello(\n invalid syntax here\n"
})
assert result.success # Write succeeds even with syntax errors
assert "lint_check" in result.data
assert result.data["lint_check"]["has_errors"]
assert "errors" in result.data["lint_check"]
def test_unicode_content(self, system_state, temp_dir):
"""Test writing Unicode content"""
tool = WriteTool(system_state)
file_path = temp_dir / "unicode.txt"
content = "Hello 世界! 🌍 Привет мир!"
result = tool.execute({
"file_path": str(file_path),
"content": content
})
assert result.success
assert file_path.read_text(encoding='utf-8') == content
def test_empty_content(self, system_state, temp_dir):
"""Test writing empty file"""
tool = WriteTool(system_state)
file_path = temp_dir / "empty.txt"
result = tool.execute({
"file_path": str(file_path),
"content": ""
})
assert result.success
assert file_path.exists()
assert file_path.read_text() == ""
def test_large_file_write(self, system_state, temp_dir):
"""Test writing large file"""
tool = WriteTool(system_state)
file_path = temp_dir / "large.txt"
content = "A" * 100000 # 100K characters
result = tool.execute({
"file_path": str(file_path),
"content": content
})
assert result.success
assert result.data["bytes_written"] == 100000
assert file_path.read_text() == content
tool_registry.py¶
"""
Tool registry - Maps tool names to implementations
"""
from typing import Dict, Type
from tools import (
BaseTool, BashTool, BashOutputTool, KillBashTool,
ReadTool, WriteTool, EditTool, MultiEditTool,
GrepTool, GlobTool, LSTool,
TodoWriteTool, ExitPlanModeTool, NotebookEditTool,
WebFetchTool, WebSearchTool, TaskTool
)
class ToolRegistry:
"""Registry of all available tools"""
def __init__(self):
self._tools: Dict[str, Type[BaseTool]] = {
"Bash": BashTool,
"BashOutput": BashOutputTool,
"KillBash": KillBashTool,
"Read": ReadTool,
"Write": WriteTool,
"Edit": EditTool,
"MultiEdit": MultiEditTool,
"Grep": GrepTool,
"Glob": GlobTool,
"LS": LSTool,
"TodoWrite": TodoWriteTool,
"ExitPlanMode": ExitPlanModeTool,
"NotebookEdit": NotebookEditTool,
"WebFetch": WebFetchTool,
"WebSearch": WebSearchTool,
"Task": TaskTool,
}
def get_tool(self, name: str, system_state) -> BaseTool:
"""Get tool instance by name"""
tool_class = self._tools.get(name)
if tool_class is None:
raise ValueError(f"Unknown tool: {name}")
return tool_class(system_state)
def get_all_tool_names(self):
"""Get list of all tool names"""
return list(self._tools.keys())
tools/__init__.py¶
"""
Tools module - All tool implementations
"""
from .base import BaseTool, ToolResult
from .bash_tool import BashTool
from .bash_output_tool import BashOutputTool
from .kill_bash_tool import KillBashTool
from .read_tool import ReadTool
from .write_tool import WriteTool
from .edit_tool import EditTool
from .multi_edit_tool import MultiEditTool
from .grep_tool import GrepTool
from .glob_tool import GlobTool
from .ls_tool import LSTool
from .todo_write_tool import TodoWriteTool
from .exit_plan_mode_tool import ExitPlanModeTool
from .notebook_edit_tool import NotebookEditTool
from .web_fetch_tool import WebFetchTool
from .web_search_tool import WebSearchTool
from .task_tool import TaskTool
from .shell_session import ShellSession
__all__ = [
'BaseTool',
'ToolResult',
'BashTool',
'BashOutputTool',
'KillBashTool',
'ReadTool',
'WriteTool',
'EditTool',
'MultiEditTool',
'GrepTool',
'GlobTool',
'LSTool',
'TodoWriteTool',
'ExitPlanModeTool',
'NotebookEditTool',
'WebFetchTool',
'WebSearchTool',
'TaskTool',
'ShellSession'
]
tools/base.py¶
"""
Base classes for tool implementation
"""
from typing import Dict, Any, Optional
from abc import ABC, abstractmethod
from dataclasses import dataclass
from datetime import datetime
@dataclass
class ToolResult:
"""Result from a tool execution"""
success: bool
data: Dict[str, Any]
error: Optional[str] = None
metadata: Optional[Dict[str, Any]] = None
def to_dict(self) -> Dict[str, Any]:
"""Convert to dictionary"""
result = self.data.copy()
if self.error:
result["error"] = self.error
if self.metadata:
result["_metadata"] = self.metadata
return result
class BaseTool(ABC):
"""Base class for all tools"""
def __init__(self, system_state: 'SystemState'):
self.state = system_state
@property
@abstractmethod
def name(self) -> str:
"""Tool name"""
pass
def execute(self, params: Dict[str, Any]) -> ToolResult:
"""
Execute the tool with given parameters
Args:
params: Tool input parameters
Returns:
ToolResult with data and metadata
"""
# Track tool call
self.state.tool_call_counts[self.name] = self.state.tool_call_counts.get(self.name, 0) + 1
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
call_number = self.state.tool_call_counts[self.name]
try:
# Call implementation
data = self._execute_impl(params)
# Add metadata
metadata = {
"tool": self.name,
"call_number": call_number,
"timestamp": timestamp
}
return ToolResult(success=True, data=data, metadata=metadata)
except Exception as e:
error_data = {
"error": str(e),
"error_type": type(e).__name__,
"tool": self.name,
"input": params
}
metadata = {
"tool": self.name,
"call_number": call_number,
"timestamp": timestamp
}
return ToolResult(success=False, data=error_data, error=str(e), metadata=metadata)
@abstractmethod
def _execute_impl(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""
Implement tool-specific logic
Args:
params: Tool input parameters
Returns:
Dictionary with tool results
"""
pass
tools/bash_output_tool.py¶
"""
BashOutput tool - Retrieve output from background bash jobs
"""
import os
import re
from typing import Dict, Any
from .base import BaseTool
class BashOutputTool(BaseTool):
"""Retrieves output from running or completed background bash shells"""
@property
def name(self) -> str:
return "BashOutput"
def _execute_impl(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""
Get output from background bash job
- Retrieves output from a running or completed background bash shell
- Takes a bash_id parameter identifying the shell
- Always returns only new output since the last check
- Supports optional regex filtering
"""
bash_id = params["bash_id"]
filter_pattern = params.get("filter")
log_file = f"/tmp/{bash_id}.log"
if not os.path.exists(log_file):
return {"error": f"Bash job not found (no output log) for bash_id: {bash_id}"}
try:
with open(log_file, 'r') as f:
output = f.read()
if filter_pattern:
# Filter lines matching pattern
lines = output.split('\n')
filtered_lines = [line for line in lines if re.search(filter_pattern, line)]
output = '\n'.join(filtered_lines)
return {
"bash_id": bash_id,
"output": output,
"output_size": len(output)
}
except Exception as e:
return {"error": f"Error reading bash output: {str(e)}"}
tools/bash_tool.py¶
"""
Bash tool - Command execution in persistent shell sessions
"""
import subprocess
import time
import hashlib
from typing import Dict, Any
from .base import BaseTool
class BashTool(BaseTool):
"""Executes bash commands in persistent shell sessions"""
@property
def name(self) -> str:
return "Bash"
def _execute_impl(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""
Execute bash command in persistent shell
- Commands execute in a persistent shell session
- Working directory changes persist across commands
- Environment variables persist
- Supports background execution with run_in_background parameter
- Output truncated if exceeds 30000 characters
"""
command = params["command"]
timeout_ms = params.get("timeout")
if timeout_ms is None:
timeout_ms = 120000
timeout = timeout_ms / 1000 # Convert ms to seconds
run_in_background = params.get("run_in_background", False)
# Get or create shell session
shell_id = self.state.default_shell_id
if shell_id not in self.state.shell_sessions:
from .shell_session import ShellSession
self.state.shell_sessions[shell_id] = ShellSession(
session_id=shell_id,
current_directory=self.state.current_directory
)
session = self.state.shell_sessions[shell_id]
if run_in_background:
# Start background process
bg_id = f"bg_{int(time.time())}_{hashlib.md5(command.encode()).hexdigest()[:8]}"
session.start()
# Launch command in background; the subshell grouping ensures the
# redirect covers the whole command (incl. && / || chains), so all
# output lands in the log file instead of leaking to this session.
bg_command = f"( {command} ) > /tmp/{bg_id}.log 2>&1 & echo $!"
output, exit_code = session.execute(bg_command, timeout=5)
return {
"output": f"Background job started with ID: {bg_id}\nPID: {output.strip()}",
"exit_code": 0,
"shell_id": shell_id,
"background_job_id": bg_id
}
else:
# Execute command synchronously
output, exit_code = session.execute(command, timeout=int(timeout))
# Update system state directory
self.state.current_directory = session.current_directory
# Truncate output if too long
if len(output) > 30000:
output = output[:30000] + f"\n... (output truncated, {len(output)} total characters)"
return {
"output": output,
"exit_code": exit_code,
"shell_id": shell_id,
"working_directory": session.current_directory
}
tools/edit_tool.py¶
"""
Edit tool - File editing with search and replace
"""
import subprocess
from pathlib import Path
from typing import Dict, Any, Optional
from .base import BaseTool
class EditTool(BaseTool):
"""Performs exact string replacements in files"""
@property
def name(self) -> str:
return "Edit"
def _execute_impl(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""
Edit file using search and replace
- You must use Read tool at least once before editing
- Ensure you preserve exact indentation (tabs/spaces)
- The edit will FAIL if old_string is not unique in the file
- Use replace_all to change every instance of old_string
"""
file_path = Path(params["file_path"]).expanduser().resolve()
old_string = params["old_string"]
new_string = params["new_string"]
replace_all = params.get("replace_all", False)
if not file_path.exists():
return {"error": f"File not found: {file_path}"}
try:
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
# Check if old_string exists
if old_string not in content:
return {"error": f"String not found in file: {old_string[:100]}..."}
# Count occurrences
occurrences = content.count(old_string)
# Check uniqueness if not replace_all
if not replace_all and occurrences > 1:
return {
"error": f"String appears {occurrences} times in file. Use replace_all=true or provide more context to make it unique."
}
# Perform replacement
if replace_all:
new_content = content.replace(old_string, new_string)
replacements = occurrences
else:
new_content = content.replace(old_string, new_string, 1)
replacements = 1
# Write back
with open(file_path, 'w', encoding='utf-8') as f:
f.write(new_content)
result = {
"file_path": str(file_path),
"replacements": replacements,
"old_length": len(content),
"new_length": len(new_content)
}
# Check for lint errors
lint_result = self._check_lint_errors(file_path)
if lint_result:
result["lint_check"] = lint_result
return result
except Exception as e:
return {"error": f"Error editing file: {str(e)}"}
def _check_lint_errors(self, file_path: Path) -> Optional[Dict[str, Any]]:
"""Check for lint errors after file modification"""
suffix = file_path.suffix
try:
if suffix == ".py":
result = subprocess.run(
["python3", "-m", "py_compile", str(file_path)],
capture_output=True,
text=True,
timeout=5
)
if result.returncode != 0:
return {
"language": "python",
"has_errors": True,
"errors": result.stderr
}
else:
return {
"language": "python",
"has_errors": False,
"message": "No syntax errors detected"
}
elif suffix in [".js", ".jsx", ".ts", ".tsx"]:
result = subprocess.run(
["node", "--check", str(file_path)],
capture_output=True,
text=True,
timeout=5
)
if result.returncode != 0:
return {
"language": "javascript/typescript",
"has_errors": True,
"errors": result.stderr
}
else:
return {
"language": "javascript/typescript",
"has_errors": False,
"message": "No syntax errors detected"
}
return None
except FileNotFoundError:
return None
except subprocess.TimeoutExpired:
return {"error": "Lint check timed out"}
except Exception:
return None
tools/exit_plan_mode_tool.py¶
"""
ExitPlanMode tool - Exit plan mode after presenting plan
"""
from typing import Dict, Any
from .base import BaseTool
class ExitPlanModeTool(BaseTool):
"""Use this tool when you are in plan mode and ready to code"""
@property
def name(self) -> str:
return "ExitPlanMode"
def _execute_impl(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""
Exit plan mode
- Use this tool when you are in plan mode and have finished presenting your plan
- This will prompt the user to exit plan mode
- IMPORTANT: Only use for tasks that require planning implementation steps for code writing
"""
plan = params["plan"]
return {
"action": "exit_plan_mode",
"plan": plan,
"message": "Plan presented. Ready to implement."
}
tools/glob_tool.py¶
"""
Glob tool - Pure Python file pattern matching
"""
import os
from pathlib import Path
from typing import Dict, Any, List
from .base import BaseTool
class GlobTool(BaseTool):
"""Fast file pattern matching tool"""
@property
def name(self) -> str:
return "Glob"
def _execute_impl(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""
Find files matching glob pattern
- Fast file pattern matching tool that works with any codebase size
- Supports glob patterns like "**/*.js" or "src/**/*.ts"
- Returns matching file paths sorted by modification time
"""
pattern = params["pattern"]
path = params.get("path", ".")
if path is None:
path = "."
# Resolve search path
search_path = Path(path).expanduser().resolve()
if not search_path.exists():
return {"error": f"Path not found: {search_path}"}
if not search_path.is_dir():
return {"error": f"Path is not a directory: {search_path}"}
# Ensure pattern starts with **/ for recursive search
if not pattern.startswith("**/"):
pattern = "**/" + pattern
# Find matching files
matches = []
try:
for match in search_path.glob(pattern):
if match.is_file():
matches.append(str(match))
except Exception as e:
return {"error": f"Error in glob search: {str(e)}"}
# Sort by modification time (newest first)
try:
matches.sort(key=lambda x: os.path.getmtime(x), reverse=True)
except Exception:
# If sorting fails, just use unsorted list
pass
return {
"pattern": pattern,
"search_path": str(search_path),
"matches": matches,
"total_matches": len(matches)
}
tools/grep_tool.py¶
"""
Grep tool - Pure Python implementation without rg/grep dependencies
Implements full regex search across files with all features from tools.json
"""
import re
import os
from pathlib import Path
from typing import Dict, Any, List, Tuple, Optional
import fnmatch
from .base import BaseTool
class GrepTool(BaseTool):
"""Pure Python grep implementation"""
@property
def name(self) -> str:
return "Grep"
def _execute_impl(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""
Search for patterns in files using pure Python regex
Supports:
- Full regex syntax
- Case insensitive search (-i)
- Context lines (-A, -B, -C)
- Line numbers (-n)
- Multiline mode
- Glob filtering
- File type filtering
- Output modes: content, files_with_matches, count
- Head limit
"""
pattern = params["pattern"]
path = params.get("path", ".")
if path is None:
path = "."
glob_pattern = params.get("glob")
output_mode = params.get("output_mode", "files_with_matches")
case_insensitive = params.get("-i", False)
context_before = params.get("-B")
if context_before is None:
context_before = 0
context_after = params.get("-A")
if context_after is None:
context_after = 0
context_around = params.get("-C")
if context_around is None:
context_around = 0
show_line_numbers = params.get("-n", False)
multiline = params.get("multiline", False)
head_limit = params.get("head_limit")
file_type = params.get("type")
# Determine context
if context_around:
context_before = context_around
context_after = context_around
# Compile regex
regex_flags = re.MULTILINE if multiline else 0
if case_insensitive:
regex_flags |= re.IGNORECASE
if multiline:
regex_flags |= re.DOTALL
try:
regex = re.compile(pattern, regex_flags)
except re.error as e:
return {"error": f"Invalid regex pattern: {str(e)}"}
# Resolve search path
search_path = Path(path).expanduser().resolve()
if not search_path.exists():
return {"error": f"Path not found: {search_path}"}
# Get files to search
files_to_search = self._get_files_to_search(
search_path, glob_pattern, file_type
)
if not files_to_search:
return {
"pattern": pattern,
"output": "No files found matching criteria.",
"matches": 0
}
# Search files
if output_mode == "files_with_matches":
results = self._search_files_with_matches(files_to_search, regex, head_limit)
elif output_mode == "count":
results = self._search_count(files_to_search, regex, head_limit)
else: # content
results = self._search_content(
files_to_search, regex,
context_before, context_after,
show_line_numbers, head_limit
)
return {
"pattern": pattern,
"output": results["output"],
"matches": results["matches"]
}
def _get_files_to_search(
self, search_path: Path, glob_pattern: Optional[str], file_type: Optional[str]
) -> List[Path]:
"""Get list of files to search"""
files = []
# Define file type extensions
type_extensions = {
"py": ["*.py"],
"python": ["*.py"],
"js": ["*.js", "*.jsx"],
"javascript": ["*.js", "*.jsx"],
"ts": ["*.ts", "*.tsx"],
"typescript": ["*.ts", "*.tsx"],
"java": ["*.java"],
"go": ["*.go"],
"rust": ["*.rs"],
"cpp": ["*.cpp", "*.cc", "*.cxx", "*.h", "*.hpp"],
"c": ["*.c", "*.h"],
"ruby": ["*.rb"],
"php": ["*.php"],
"html": ["*.html", "*.htm"],
"css": ["*.css"],
"json": ["*.json"],
"yaml": ["*.yaml", "*.yml"],
"md": ["*.md"],
"markdown": ["*.md"],
"txt": ["*.txt"],
}
if search_path.is_file():
# Single file
files = [search_path]
else:
# Directory - walk recursively
for root, dirs, filenames in os.walk(search_path):
# Skip hidden directories
dirs[:] = [d for d in dirs if not d.startswith('.')]
for filename in filenames:
# Skip hidden files
if filename.startswith('.'):
continue
file_path = Path(root) / filename
# Check file type filter
if file_type:
extensions = type_extensions.get(file_type, [])
if not any(fnmatch.fnmatch(filename, ext) for ext in extensions):
continue
# Check glob filter
if glob_pattern:
# Convert glob to relative path for matching
try:
rel_path = file_path.relative_to(search_path)
if not fnmatch.fnmatch(str(rel_path), glob_pattern):
continue
except ValueError:
continue
files.append(file_path)
return files
def _search_files_with_matches(
self, files: List[Path], regex: re.Pattern, head_limit: Optional[int]
) -> Dict[str, Any]:
"""Search and return files that have matches"""
matching_files = []
for file_path in files:
try:
# Try to read as text
with open(file_path, 'r', encoding='utf-8', errors='ignore') as f:
content = f.read()
# Check for match
if regex.search(content):
matching_files.append(str(file_path))
# Check head limit
if head_limit and len(matching_files) >= head_limit:
break
except (IOError, OSError, UnicodeDecodeError):
# Skip files that can't be read
continue
if not matching_files:
output = "No matches found."
else:
output = "\n".join(matching_files)
return {
"output": output,
"matches": len(matching_files)
}
def _search_count(
self, files: List[Path], regex: re.Pattern, head_limit: Optional[int]
) -> Dict[str, Any]:
"""Count matches per file"""
results = []
total_matches = 0
for file_path in files:
try:
with open(file_path, 'r', encoding='utf-8', errors='ignore') as f:
content = f.read()
# Count matches
matches = len(regex.findall(content))
if matches > 0:
results.append(f"{file_path}:{matches}")
total_matches += matches
if head_limit and len(results) >= head_limit:
break
except (IOError, OSError, UnicodeDecodeError):
continue
if not results:
output = "No matches found."
else:
output = "\n".join(results)
return {
"output": output,
"matches": total_matches
}
def _search_content(
self, files: List[Path], regex: re.Pattern,
context_before: int, context_after: int,
show_line_numbers: bool, head_limit: Optional[int]
) -> Dict[str, Any]:
"""Search and return matching lines with context"""
output_lines = []
total_matches = 0
lines_added = 0
for file_path in files:
try:
with open(file_path, 'r', encoding='utf-8', errors='ignore') as f:
lines = f.readlines()
# Find matching lines
matching_line_numbers = []
for i, line in enumerate(lines):
if regex.search(line):
matching_line_numbers.append(i)
if not matching_line_numbers:
continue
# Add file header
output_lines.append(f"\n{file_path}")
lines_added += 1
# Process each match with context
lines_to_show = set()
for line_num in matching_line_numbers:
# Add context lines
start = max(0, line_num - context_before)
end = min(len(lines), line_num + context_after + 1)
lines_to_show.update(range(start, end))
# Output lines in order
prev_line = -2
for line_num in sorted(lines_to_show):
# Add separator for gaps
if line_num > prev_line + 1:
output_lines.append("--")
lines_added += 1
line = lines[line_num].rstrip()
is_match = line_num in matching_line_numbers
# Format line
if show_line_numbers:
prefix = f"{line_num + 1}:"
else:
prefix = ""
# Use : for match lines, - for context
separator = ":" if is_match else "-"
formatted = f"{prefix}{separator}{line}" if prefix else f"{separator}{line}"
output_lines.append(formatted)
lines_added += 1
if is_match:
total_matches += 1
prev_line = line_num
# Check head limit
if head_limit and lines_added >= head_limit:
break
if head_limit and lines_added >= head_limit:
break
except (IOError, OSError, UnicodeDecodeError):
continue
if not output_lines:
output = "No matches found."
else:
output = "\n".join(output_lines)
return {
"output": output,
"matches": total_matches
}
tools/kill_bash_tool.py¶
"""
KillBash tool - Terminate shell sessions
"""
from typing import Dict, Any
from .base import BaseTool
class KillBashTool(BaseTool):
"""Kills a running background bash shell by its ID"""
@property
def name(self) -> str:
return "KillBash"
def _execute_impl(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""
Kill a shell session
- Kills a running background bash shell by its ID
- Takes a shell_id parameter identifying the shell to kill
- Returns a success or failure status
"""
shell_id = params["shell_id"]
if shell_id not in self.state.shell_sessions:
return {"error": f"Shell session not found: {shell_id}"}
try:
session = self.state.shell_sessions[shell_id]
session.kill()
del self.state.shell_sessions[shell_id]
return {
"shell_id": shell_id,
"status": "terminated"
}
except Exception as e:
return {"error": f"Error killing shell: {str(e)}"}
tools/ls_tool.py¶
"""
LS tool - Directory listing
"""
import os
from pathlib import Path
import fnmatch
from typing import Dict, Any, List
from .base import BaseTool
class LSTool(BaseTool):
"""Lists files and directories"""
@property
def name(self) -> str:
return "LS"
def _execute_impl(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""
List directory contents
- The path parameter must be an absolute path
- You can optionally provide an array of glob patterns to ignore
"""
path = Path(params["path"]).expanduser().resolve()
ignore_patterns = params.get("ignore")
if ignore_patterns is None:
ignore_patterns = []
if not path.exists():
return {"error": f"Path not found: {path}"}
if not path.is_dir():
return {"error": f"Not a directory: {path}"}
try:
entries = []
for entry in sorted(path.iterdir()):
# Skip hidden files (starting with .)
if entry.name.startswith('.'):
continue
# Check ignore patterns
should_ignore = False
for pattern in ignore_patterns:
if fnmatch.fnmatch(entry.name, pattern):
should_ignore = True
break
if should_ignore:
continue
# Get entry info
entry_type = "dir" if entry.is_dir() else "file"
size = entry.stat().st_size if entry.is_file() else 0
entries.append({
"name": entry.name,
"type": entry_type,
"size": size,
"path": str(entry)
})
return {
"path": str(path),
"entries": entries,
"total_entries": len(entries)
}
except PermissionError:
return {"error": f"Permission denied: {path}"}
except Exception as e:
return {"error": f"Error listing directory: {str(e)}"}
tools/multi_edit_tool.py¶
"""
MultiEdit tool - Multiple edits to a single file in one operation
"""
import subprocess
from pathlib import Path
from typing import Dict, Any, List, Optional
from .base import BaseTool
class MultiEditTool(BaseTool):
"""Makes multiple edits to a single file in one operation"""
@property
def name(self) -> str:
return "MultiEdit"
def _execute_impl(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""
Perform multiple edits on a file
- Built on top of Edit tool
- All edits are applied in sequence, in the order they are provided
- Each edit operates on the result of the previous edit
- All edits must be valid for the operation to succeed - if any edit fails, none will be applied
- The edits are atomic - either all succeed or none are applied
"""
file_path = Path(params["file_path"]).expanduser().resolve()
edits = params.get("edits")
if edits is None:
edits = []
if not file_path.exists():
# Check if this is a file creation (first edit has empty old_string)
if edits and edits[0]["old_string"] == "":
try:
file_path.parent.mkdir(parents=True, exist_ok=True)
with open(file_path, 'w', encoding='utf-8') as f:
f.write("")
except Exception as e:
return {"error": f"Error creating file: {str(e)}"}
else:
return {"error": f"File not found: {file_path}"}
try:
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
original_content = content
results = []
# Apply edits sequentially
for i, edit in enumerate(edits):
old_string = edit["old_string"]
new_string = edit["new_string"]
replace_all = edit.get("replace_all", False)
if old_string == "" and i == 0:
# File creation case
content = new_string
results.append({"edit": i + 1, "action": "created", "success": True})
continue
if old_string not in content:
return {
"error": f"Edit #{i + 1} failed: String not found",
"old_string": old_string[:100],
"completed_edits": i
}
occurrences = content.count(old_string)
if not replace_all and occurrences > 1:
return {
"error": f"Edit #{i + 1} failed: String appears {occurrences} times",
"completed_edits": i
}
if replace_all:
content = content.replace(old_string, new_string)
replacements = occurrences
else:
content = content.replace(old_string, new_string, 1)
replacements = 1
results.append({
"edit": i + 1,
"replacements": replacements,
"success": True
})
# Write back
with open(file_path, 'w', encoding='utf-8') as f:
f.write(content)
result = {
"file_path": str(file_path),
"total_edits": len(edits),
"successful_edits": len(results),
"edit_results": results,
"old_size": len(original_content),
"new_size": len(content)
}
# Check for lint errors
lint_result = self._check_lint_errors(file_path)
if lint_result:
result["lint_check"] = lint_result
return result
except Exception as e:
return {"error": f"Error in multi-edit: {str(e)}"}
def _check_lint_errors(self, file_path: Path) -> Optional[Dict[str, Any]]:
"""Check for lint errors"""
suffix = file_path.suffix
try:
if suffix == ".py":
result = subprocess.run(
["python3", "-m", "py_compile", str(file_path)],
capture_output=True, text=True, timeout=5
)
return {
"language": "python",
"has_errors": result.returncode != 0,
"errors": result.stderr if result.returncode != 0 else None,
"message": "No syntax errors detected" if result.returncode == 0 else None
}
return None
except:
return None
tools/notebook_edit_tool.py¶
"""
NotebookEdit tool - Edit Jupyter notebook cells
"""
import json
from pathlib import Path
from typing import Dict, Any
from .base import BaseTool
class NotebookEditTool(BaseTool):
"""Completely replaces the contents of a specific cell in a Jupyter notebook"""
@property
def name(self) -> str:
return "NotebookEdit"
def _execute_impl(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""
Edit Jupyter notebook cell
- Completely replaces the contents of a specific cell
- The notebook_path parameter must be an absolute path
- Use edit_mode=insert to add a new cell
- Use edit_mode=delete to delete a cell
- Use edit_mode=replace to replace cell contents (default)
"""
notebook_path = Path(params["notebook_path"]).expanduser().resolve()
cell_id = params.get("cell_id")
new_source = params.get("new_source")
cell_type = params.get("cell_type", "code")
edit_mode = params.get("edit_mode", "replace")
if not notebook_path.exists():
return {"error": f"Notebook not found: {notebook_path}"}
try:
# Load notebook
with open(notebook_path, 'r', encoding='utf-8') as f:
notebook = json.load(f)
cells = notebook.get('cells', [])
if edit_mode == "insert":
if new_source is None:
return {"error": "new_source required for insert mode"}
# Insert new cell
new_cell = {
"cell_type": cell_type,
"metadata": {},
"source": new_source.split('\n')
}
if cell_type == "code":
new_cell["outputs"] = []
new_cell["execution_count"] = None
# Find insertion point
if cell_id:
# Insert after cell with given ID
for i, cell in enumerate(cells):
if cell.get('id') == cell_id:
cells.insert(i + 1, new_cell)
break
else:
return {"error": f"Cell with ID {cell_id} not found"}
else:
# Insert at beginning
cells.insert(0, new_cell)
action = "inserted"
elif edit_mode == "delete":
# Delete cell
if cell_id:
for i, cell in enumerate(cells):
if cell.get('id') == cell_id:
cells.pop(i)
break
else:
return {"error": f"Cell with ID {cell_id} not found"}
else:
return {"error": "cell_id required for delete mode"}
action = "deleted"
else: # replace
if new_source is None:
return {"error": "new_source required for replace mode"}
# Replace cell contents
if cell_id:
for cell in cells:
if cell.get('id') == cell_id:
cell["source"] = new_source.split('\n')
if cell_type:
cell["cell_type"] = cell_type
break
else:
return {"error": f"Cell with ID {cell_id} not found"}
else:
return {"error": "cell_id required for replace mode"}
action = "replaced"
# Update notebook
notebook["cells"] = cells
# Write back
with open(notebook_path, 'w', encoding='utf-8') as f:
json.dump(notebook, f, indent=1, ensure_ascii=False)
return {
"notebook_path": str(notebook_path),
"action": action,
"total_cells": len(cells)
}
except json.JSONDecodeError:
return {"error": "Invalid Jupyter notebook format"}
except Exception as e:
return {"error": f"Error editing notebook: {str(e)}"}
tools/read_tool.py¶
"""
Read tool - File reading with support for text, images, PDFs, and notebooks
"""
import os
from pathlib import Path
from typing import Dict, Any, Optional
from .base import BaseTool
class ReadTool(BaseTool):
"""Reads files from the local filesystem"""
@property
def name(self) -> str:
return "Read"
def _execute_impl(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""
Read file contents
- The file_path parameter must be an absolute path
- By default, reads up to 2000 lines from the beginning
- Can specify offset and limit for large files
- Lines longer than 2000 characters are truncated
- Results returned in cat -n format with line numbers starting at 1
- Supports images, PDFs, Jupyter notebooks
"""
file_path = Path(params["file_path"]).expanduser().resolve()
offset = params.get("offset")
if offset is None:
offset = 0
limit = params.get("limit")
if limit is None:
limit = 2000
if not file_path.exists():
return {"error": f"File not found: {file_path}"}
if not file_path.is_file():
return {"error": f"Not a file: {file_path}"}
# Check file type
suffix = file_path.suffix.lower()
# Handle special file types
if suffix in ['.png', '.jpg', '.jpeg', '.gif', '.bmp', '.webp']:
return self._read_image(file_path)
elif suffix == '.pdf':
return self._read_pdf(file_path)
elif suffix == '.ipynb':
return self._read_notebook(file_path)
else:
return self._read_text(file_path, offset, limit)
def _read_text(self, file_path: Path, offset: int, limit: int) -> Dict[str, Any]:
"""Read text file"""
try:
# Sniff for binary content first: NUL bytes never appear in text,
# and control bytes like \x00-\x05 are valid UTF-8, so a decode
# error alone is not a reliable binary signal.
with open(file_path, 'rb') as f:
sample = f.read(8192)
if b'\x00' in sample:
return {"error": "File appears to be binary. Cannot read as text."}
with open(file_path, 'r', encoding='utf-8') as f:
lines = f.readlines()
# Apply offset and limit
total_lines = len(lines)
selected_lines = lines[offset:offset + limit] if offset or limit < total_lines else lines
# Format with line numbers (1-indexed)
formatted_lines = []
for i, line in enumerate(selected_lines, start=offset + 1):
# Truncate long lines
line_content = line.rstrip()
if len(line_content) > 2000:
line_content = line_content[:2000] + "... (line truncated)"
formatted_lines.append(f"{i:6d}|{line_content}")
content = "\n".join(formatted_lines)
if not content:
content = "File is empty."
return {
"file_path": str(file_path),
"total_lines": total_lines,
"showing_lines": f"{offset + 1}-{offset + len(selected_lines)}",
"content": content
}
except UnicodeDecodeError:
return {"error": "File appears to be binary. Cannot read as text."}
except Exception as e:
return {"error": f"Error reading file: {str(e)}"}
def _read_image(self, file_path: Path) -> Dict[str, Any]:
"""Read image file"""
# For now, just return metadata since we can't display images in text
try:
size = file_path.stat().st_size
return {
"file_path": str(file_path),
"file_type": "image",
"format": file_path.suffix[1:].upper(),
"size_bytes": size,
"note": "Image file detected. Full visual analysis requires multimodal LLM support."
}
except Exception as e:
return {"error": f"Error reading image: {str(e)}"}
def _read_pdf(self, file_path: Path) -> Dict[str, Any]:
"""Read PDF file"""
# For now, return basic info
# Full PDF support would require PyPDF2 or similar
try:
size = file_path.stat().st_size
return {
"file_path": str(file_path),
"file_type": "pdf",
"size_bytes": size,
"note": "PDF file detected. Full text extraction requires PyPDF2 library. Install with: pip install PyPDF2"
}
except Exception as e:
return {"error": f"Error reading PDF: {str(e)}"}
def _read_notebook(self, file_path: Path) -> Dict[str, Any]:
"""Read Jupyter notebook"""
try:
import json
with open(file_path, 'r', encoding='utf-8') as f:
notebook = json.load(f)
# Extract cells
cells = notebook.get('cells', [])
# Format output
output_lines = []
output_lines.append(f"Jupyter Notebook: {file_path.name}")
output_lines.append("=" * 60)
for i, cell in enumerate(cells):
cell_type = cell.get('cell_type', 'unknown')
source = cell.get('source', [])
# Convert source to string
if isinstance(source, list):
source_text = ''.join(source)
else:
source_text = source
output_lines.append(f"\n[Cell {i + 1}] Type: {cell_type}")
output_lines.append("-" * 60)
output_lines.append(source_text)
# Show outputs for code cells
if cell_type == 'code':
outputs = cell.get('outputs', [])
if outputs:
output_lines.append("\nOutput:")
for output in outputs:
output_type = output.get('output_type', '')
if output_type == 'stream':
text = ''.join(output.get('text', []))
output_lines.append(text)
elif output_type == 'execute_result' or output_type == 'display_data':
data = output.get('data', {})
if 'text/plain' in data:
text = ''.join(data['text/plain'])
output_lines.append(text)
content = '\n'.join(output_lines)
return {
"file_path": str(file_path),
"file_type": "jupyter_notebook",
"total_cells": len(cells),
"content": content
}
except json.JSONDecodeError:
return {"error": "Invalid Jupyter notebook format"}
except Exception as e:
return {"error": f"Error reading notebook: {str(e)}"}
tools/shell_session.py¶
"""
Shell session management for persistent bash execution
"""
import subprocess
import time
import os
import re
import queue
import threading
from dataclasses import dataclass, field
from typing import Optional, Tuple, Dict
@dataclass
class ShellSession:
"""Manages a persistent shell session"""
session_id: str
process: Optional[subprocess.Popen] = None
current_directory: str = field(default_factory=lambda: os.getcwd())
env: Dict[str, str] = field(default_factory=lambda: os.environ.copy())
output_buffer: str = ""
def start(self):
"""Start the shell process"""
if self.process is None or self.process.poll() is not None:
self.process = subprocess.Popen(
["/bin/bash"],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
bufsize=1,
cwd=self.current_directory,
env=self.env
)
# Reader thread feeds stdout lines into a queue so that execute()
# can wait on them with a real timeout (a bare readline() would
# block forever on silent commands like `sleep`).
self._output_queue = queue.Queue()
reader = threading.Thread(
target=self._read_stdout,
args=(self.process, self._output_queue),
daemon=True
)
reader.start()
def _read_stdout(self, process, q):
"""Pump stdout lines into the queue; None marks end of stream."""
for line in iter(process.stdout.readline, ''):
q.put(line)
q.put(None)
def _restart(self):
"""Kill a stuck shell and start a fresh one (keeps current_directory)."""
self.kill()
self.process = None
self.start()
def execute(self, command: str, timeout: int = 120) -> Tuple[str, int]:
"""Execute command in the persistent shell"""
self.start()
try:
# Send command
full_command = f"cd {self.current_directory} && {command}\n"
self.process.stdin.write(full_command)
self.process.stdin.write("echo __CMD_DONE__$?\n")
self.process.stdin.flush()
# Read output until marker
output_lines = []
deadline = time.time() + timeout
exit_code = -1
while True:
remaining = deadline - time.time()
if remaining <= 0:
# Kill the stuck shell so the pending marker of the
# timed-out command cannot corrupt the next command.
self._restart()
return f"Command timed out (timeout: {timeout}s)", -1
try:
line = self._output_queue.get(timeout=remaining)
except queue.Empty:
self._restart()
return f"Command timed out (timeout: {timeout}s)", -1
if line is None: # shell exited unexpectedly
break
# 只认独占一行的结束标记；命令输出中恰好包含标记字符串的行
# 视为普通输出，避免误截断或与下一命令的输出错位。
m = re.fullmatch(r"__CMD_DONE__(\d+)", line.strip())
if m:
exit_code = int(m.group(1))
break
output_lines.append(line.rstrip())
output = "\n".join(output_lines)
# Update current directory if cd was used
if command.strip().startswith("cd "):
try:
self.process.stdin.write("pwd\n")
self.process.stdin.flush()
new_dir = self._output_queue.get(timeout=5)
if new_dir:
new_dir = new_dir.strip()
if new_dir and os.path.isdir(new_dir):
self.current_directory = new_dir
except Exception:
pass
return output, exit_code
except Exception as e:
return f"Error executing command: {str(e)}", -1
def kill(self):
"""Terminate the shell session"""
if self.process and self.process.poll() is None:
self.process.terminate()
try:
self.process.wait(timeout=5)
except subprocess.TimeoutExpired:
self.process.kill()
tools/task_tool.py¶
"""
Task tool - Launch sub-agents for complex tasks
"""
from typing import Dict, Any
from .base import BaseTool
class TaskTool(BaseTool):
"""Launch a new agent to handle complex, multi-step tasks autonomously"""
@property
def name(self) -> str:
return "Task"
def _execute_impl(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""
Launch sub-agent
- Launch a new agent to handle complex, multi-step tasks autonomously
- Available agent types: general-purpose, statusline-setup, output-style-setup
- NOTE: This is a stub implementation. Full implementation would require:
- Recursive agent instantiation
- Isolated execution context
- Result aggregation
"""
description = params["description"]
prompt = params["prompt"]
subagent_type = params["subagent_type"]
return {
"description": description,
"subagent_type": subagent_type,
"error": "Task tool (sub-agents) not yet implemented",
"note": "This tool would launch a specialized sub-agent to handle the task autonomously"
}
tools/todo_write_tool.py¶
"""
TodoWrite tool - Task list management
"""
from typing import Dict, Any
from .base import BaseTool
class TodoWriteTool(BaseTool):
"""Creates and manages structured task lists"""
@property
def name(self) -> str:
return "TodoWrite"
def _execute_impl(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""
Update TODO list
- Use this tool to create and manage a structured task list
- Track progress, organize complex tasks
- Helps user understand progress
"""
# JSON null for required todos: same as empty list (LLM omit-as-null).
todos = params["todos"]
if todos is None:
todos = []
# Validate todo format
for todo in todos:
if not all(k in todo for k in ["id", "content", "status"]):
return {"error": "Each todo must have id, content, and status"}
if todo["status"] not in ["pending", "in_progress", "completed"]:
return {"error": f"Invalid status: {todo['status']}"}
# Update state
self.state.todos = todos
return {
"total_todos": len(todos),
"pending": sum(1 for t in todos if t["status"] == "pending"),
"in_progress": sum(1 for t in todos if t["status"] == "in_progress"),
"completed": sum(1 for t in todos if t["status"] == "completed")
}
tools/web_fetch_tool.py¶
"""
WebFetch tool - Fetch and analyze web content
"""
from typing import Dict, Any
from .base import BaseTool
class WebFetchTool(BaseTool):
"""Fetches content from a specified URL and processes it"""
@property
def name(self) -> str:
return "WebFetch"
def _execute_impl(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""
Fetch content from URL
- Fetches content from a specified URL and processes it using an AI model
- Takes a URL and a prompt as input
- Fetches the URL content, converts HTML to markdown
- Returns the model's response about the content
- NOTE: This is a stub implementation. Full implementation would require:
- requests library for HTTP
- beautifulsoup4 for HTML parsing
- html2text for markdown conversion
"""
url = params["url"]
prompt = params["prompt"]
return {
"url": url,
"error": "WebFetch tool requires additional dependencies. Install with: pip install requests beautifulsoup4 html2text",
"note": "This tool would fetch web content and process it with the given prompt"
}
tools/web_search_tool.py¶
"""
WebSearch tool - Search the web
"""
from typing import Dict, Any
from .base import BaseTool
class WebSearchTool(BaseTool):
"""Allows Claude to search the web and use the results"""
@property
def name(self) -> str:
return "WebSearch"
def _execute_impl(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""
Search the web
- Allows Claude to search the web and use the results to inform responses
- Provides up-to-date information for current events and recent data
- NOTE: This is a stub implementation. Full implementation would require
API integration with search services like Google, Bing, or DuckDuckGo
"""
query = params["query"]
allowed_domains = params.get("allowed_domains", [])
blocked_domains = params.get("blocked_domains", [])
return {
"query": query,
"error": "WebSearch tool requires API integration with a search service",
"note": "This tool would search the web with the given query and filters"
}
tools/write_tool.py¶
"""
Write tool - File writing with automatic lint checking
"""
import subprocess
from pathlib import Path
from typing import Dict, Any, Optional
from .base import BaseTool
class WriteTool(BaseTool):
"""Writes files to the local filesystem"""
@property
def name(self) -> str:
return "Write"
def _execute_impl(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""
Write content to file
- This tool will overwrite the existing file if there is one at the provided path
- ALWAYS prefer editing existing files in the codebase
- NEVER write new files unless explicitly required
- NEVER proactively create documentation files (*.md) or README files
"""
file_path = Path(params["file_path"]).expanduser().resolve()
content = params["content"]
try:
# Create parent directories if needed
file_path.parent.mkdir(parents=True, exist_ok=True)
# Write file
with open(file_path, 'w', encoding='utf-8') as f:
f.write(content)
result = {
"file_path": str(file_path),
"bytes_written": len(content.encode('utf-8')),
"lines_written": len(content.split('\n'))
}
# Check for lint errors
lint_result = self._check_lint_errors(file_path)
if lint_result:
result["lint_check"] = lint_result
return result
except Exception as e:
return {"error": f"Error writing file: {str(e)}"}
def _check_lint_errors(self, file_path: Path) -> Optional[Dict[str, Any]]:
"""Check for lint errors after file modification"""
suffix = file_path.suffix
try:
if suffix == ".py":
# Check Python syntax
result = subprocess.run(
["python3", "-m", "py_compile", str(file_path)],
capture_output=True,
text=True,
timeout=5
)
if result.returncode != 0:
return {
"language": "python",
"has_errors": True,
"errors": result.stderr
}
else:
return {
"language": "python",
"has_errors": False,
"message": "No syntax errors detected"
}
elif suffix in [".js", ".jsx", ".ts", ".tsx"]:
# Check JavaScript/TypeScript with node if available
result = subprocess.run(
["node", "--check", str(file_path)],
capture_output=True,
text=True,
timeout=5
)
if result.returncode != 0:
return {
"language": "javascript/typescript",
"has_errors": True,
"errors": result.stderr
}
else:
return {
"language": "javascript/typescript",
"has_errors": False,
"message": "No syntax errors detected"
}
# No linter available for this file type
return None
except FileNotFoundError:
# Linter not installed
return None
except subprocess.TimeoutExpired:
return {"error": "Lint check timed out"}
except Exception:
return None
tools.json¶
{
"tools": [
{
"name": "Task",
"description": "Launch a new agent to handle complex, multi-step tasks autonomously. \n\nAvailable agent types and the tools they have access to:\n- general-purpose: General-purpose agent for researching complex questions, searching for code, and executing multi-step tasks. When you are searching for a keyword or file and are not confident that you will find the right match in the first few tries use this agent to perform the search for you. (Tools: *)\n- statusline-setup: Use this agent to configure the user's Claude Code status line setting. (Tools: Read, Edit)\n- output-style-setup: Use this agent to create a Claude Code output style. (Tools: Read, Write, Edit, Glob, LS, Grep)\n\nWhen using the Task tool, you must specify a subagent_type parameter to select which agent type to use.\n\n\n\nWhen NOT to use the Agent tool:\n- If you want to read a specific file path, use the Read or Glob tool instead of the Agent tool, to find the match more quickly\n- If you are searching for a specific class definition like \"class Foo\", use the Glob tool instead, to find the match more quickly\n- If you are searching for code within a specific file or set of 2-3 files, use the Read tool instead of the Agent tool, to find the match more quickly\n- Other tasks that are not related to the agent descriptions above\n\n\nUsage notes:\n1. Launch multiple agents concurrently whenever possible, to maximize performance; to do that, use a single message with multiple tool uses\n2. When the agent is done, it will return a single message back to you. The result returned by the agent is not visible to the user. To show the user the result, you should send a text message back to the user with a concise summary of the result.\n3. Each agent invocation is stateless. You will not be able to send additional messages to the agent, nor will the agent be able to communicate with you outside of its final report. Therefore, your prompt should contain a highly detailed task description for the agent to perform autonomously and you should specify exactly what information the agent should return back to you in its final and only message to you.\n4. The agent's outputs should generally be trusted\n5. Clearly tell the agent whether you expect it to write code or just to do research (search, file reads, web fetches, etc.), since it is not aware of the user's intent\n6. If the agent description mentions that it should be used proactively, then you should try your best to use it without the user having to ask for it first. Use your judgement.\n\nExample usage:\n\n<example_agent_descriptions>\n\"code-reviewer\": use this agent after you are done writing a signficant piece of code\n\"greeting-responder\": use this agent when to respond to user greetings with a friendly joke\n</example_agent_description>\n\n<example>\nuser: \"Please write a function that checks if a number is prime\"\nassistant: Sure let me write a function that checks if a number is prime\nassistant: First let me use the Write tool to write a function that checks if a number is prime\nassistant: I'm going to use the Write tool to write the following code:\n<code>\nfunction isPrime(n) {\n if (n <= 1) return false\n for (let i = 2; i * i <= n; i++) {\n if (n % i === 0) return false\n }\n return true\n}\n</code>\n<commentary>\nSince a signficant piece of code was written and the task was completed, now use the code-reviewer agent to review the code\n</commentary>\nassistant: Now let me use the code-reviewer agent to review the code\nassistant: Uses the Task tool to launch the with the code-reviewer agent \n</example>\n\n<example>\nuser: \"Hello\"\n<commentary>\nSince the user is greeting, use the greeting-responder agent to respond with a friendly joke\n</commentary>\nassistant: \"I'm going to use the Task tool to launch the with the greeting-responder agent\"\n</example>\n",
"input_schema": {
"type": "object",
"properties": {
"description": {
"type": "string",
"description": "A short (3-5 word) description of the task"
},
"prompt": {
"type": "string",
"description": "The task for the agent to perform"
},
"subagent_type": {
"type": "string",
"description": "The type of specialized agent to use for this task"
}
},
"required": [
"description",
"prompt",
"subagent_type"
],
"additionalProperties": false,
"$schema": "http://json-schema.org/draft-07/schema#"
}
},
{
"name": "Bash",
"description": "Executes a given bash command in a persistent shell session with optional timeout, ensuring proper handling and security measures.\n\nBefore executing the command, please follow these steps:\n\n1. Directory Verification:\n - If the command will create new directories or files, first use the LS tool to verify the parent directory exists and is the correct location\n - For example, before running \"mkdir foo/bar\", first use LS to check that \"foo\" exists and is the intended parent directory\n\n2. Command Execution:\n - Always quote file paths that contain spaces with double quotes (e.g., cd \"path with spaces/file.txt\")\n - Examples of proper quoting:\n - cd \"/Users/name/My Documents\" (correct)\n - cd /Users/name/My Documents (incorrect - will fail)\n - python \"/path/with spaces/script.py\" (correct)\n - python /path/with spaces/script.py (incorrect - will fail)\n - After ensuring proper quoting, execute the command.\n - Capture the output of the command.\n\nUsage notes:\n - The command argument is required.\n - You can specify an optional timeout in milliseconds (up to 600000ms / 10 minutes). If not specified, commands will timeout after 120000ms (2 minutes).\n - It is very helpful if you write a clear, concise description of what this command does in 5-10 words.\n - If the output exceeds 30000 characters, output will be truncated before being returned to you.\n - You can use the `run_in_background` parameter to run the command in the background, which allows you to continue working while the command runs. You can monitor the output using the Bash tool as it becomes available. Never use `run_in_background` to run 'sleep' as it will return immediately. You do not need to use '&' at the end of the command when using this parameter.\n - VERY IMPORTANT: You MUST avoid using search commands like `find` and `grep`. Instead use Grep, Glob, or Task to search. You MUST avoid read tools like `cat`, `head`, `tail`, and `ls`, and use Read and LS to read files.\n - If you _still_ need to run `grep`, STOP. ALWAYS USE ripgrep at `rg` first, which all Claude Code users have pre-installed.\n - When issuing multiple commands, use the ';' or '&&' operator to separate them. DO NOT use newlines (newlines are ok in quoted strings).\n - Try to maintain your current working directory throughout the session by using absolute paths and avoiding usage of `cd`. You may use `cd` if the User explicitly requests it.\n <good-example>\n pytest /foo/bar/tests\n </good-example>\n <bad-example>\n cd /foo/bar && pytest tests\n </bad-example>\n\n\n# Committing changes with git\n\nWhen the user asks you to create a new git commit, follow these steps carefully:\n\n1. You have the capability to call multiple tools in a single response. When multiple independent pieces of information are requested, batch your tool calls together for optimal performance. ALWAYS run the following bash commands in parallel, each using the Bash tool:\n - Run a git status command to see all untracked files.\n - Run a git diff command to see both staged and unstaged changes that will be committed.\n - Run a git log command to see recent commit messages, so that you can follow this repository's commit message style.\n2. Analyze all staged changes (both previously staged and newly added) and draft a commit message:\n - Summarize the nature of the changes (eg. new feature, enhancement to an existing feature, bug fix, refactoring, test, docs, etc.). Ensure the message accurately reflects the changes and their purpose (i.e. \"add\" means a wholly new feature, \"update\" means an enhancement to an existing feature, \"fix\" means a bug fix, etc.).\n - Check for any sensitive information that shouldn't be committed\n - Draft a concise (1-2 sentences) commit message that focuses on the \"why\" rather than the \"what\"\n - Ensure it accurately reflects the changes and their purpose\n3. You have the capability to call multiple tools in a single response. When multiple independent pieces of information are requested, batch your tool calls together for optimal performance. ALWAYS run the following commands in parallel:\n - Add relevant untracked files to the staging area.\n - Create the commit with a message ending with:\n 🤖 Generated with [Claude Code](https://claude.ai/code)\n\n Co-Authored-By: Claude <noreply@anthropic.com>\n - Run git status to make sure the commit succeeded.\n4. If the commit fails due to pre-commit hook changes, retry the commit ONCE to include these automated changes. If it fails again, it usually means a pre-commit hook is preventing the commit. If the commit succeeds but you notice that files were modified by the pre-commit hook, you MUST amend your commit to include them.\n\nImportant notes:\n- NEVER update the git config\n- NEVER run additional commands to read or explore code, besides git bash commands\n- NEVER use the TodoWrite or Task tools\n- DO NOT push to the remote repository unless the user explicitly asks you to do so\n- IMPORTANT: Never use git commands with the -i flag (like git rebase -i or git add -i) since they require interactive input which is not supported.\n- If there are no changes to commit (i.e., no untracked files and no modifications), do not create an empty commit\n- In order to ensure good formatting, ALWAYS pass the commit message via a HEREDOC, a la this example:\n<example>\ngit commit -m \"$(cat <<'EOF'\n Commit message here.\n\n 🤖 Generated with [Claude Code](https://claude.ai/code)\n\n Co-Authored-By: Claude <noreply@anthropic.com>\n EOF\n )\"\n</example>\n\n# Creating pull requests\nUse the gh command via the Bash tool for ALL GitHub-related tasks including working with issues, pull requests, checks, and releases. If given a Github URL use the gh command to get the information needed.\n\nIMPORTANT: When the user asks you to create a pull request, follow these steps carefully:\n\n1. You have the capability to call multiple tools in a single response. When multiple independent pieces of information are requested, batch your tool calls together for optimal performance. ALWAYS run the following bash commands in parallel using the Bash tool, in order to understand the current state of the branch since it diverged from the main branch:\n - Run a git status command to see all untracked files\n - Run a git diff command to see both staged and unstaged changes that will be committed\n - Check if the current branch tracks a remote branch and is up to date with the remote, so you know if you need to push to the remote\n - Run a git log command and `git diff [base-branch]...HEAD` to understand the full commit history for the current branch (from the time it diverged from the base branch)\n2. Analyze all changes that will be included in the pull request, making sure to look at all relevant commits (NOT just the latest commit, but ALL commits that will be included in the pull request!!!), and draft a pull request summary\n3. You have the capability to call multiple tools in a single response. When multiple independent pieces of information are requested, batch your tool calls together for optimal performance. ALWAYS run the following commands in parallel:\n - Create new branch if needed\n - Push to remote with -u flag if needed\n - Create PR using gh pr create with the format below. Use a HEREDOC to pass the body to ensure correct formatting.\n<example>\ngh pr create --title \"the pr title\" --body \"$(cat <<'EOF'\n## Summary\n<1-3 bullet points>\n\n## Test plan\n[Checklist of TODOs for testing the pull request...]\n\n🤖 Generated with [Claude Code](https://claude.ai/code)\nEOF\n)\"\n</example>\n\nImportant:\n- NEVER update the git config\n- DO NOT use the TodoWrite or Task tools\n- Return the PR URL when you're done, so the user can see it\n\n# Other common operations\n- View comments on a Github PR: gh api repos/foo/bar/pulls/123/comments",
"input_schema": {
"type": "object",
"properties": {
"command": {
"type": "string",
"description": "The command to execute"
},
"timeout": {
"type": "number",
"description": "Optional timeout in milliseconds (max 600000)"
},
"description": {
"type": "string",
"description": " Clear, concise description of what this command does in 5-10 words. Examples:\nInput: ls\nOutput: Lists files in current directory\n\nInput: git status\nOutput: Shows working tree status\n\nInput: npm install\nOutput: Installs package dependencies\n\nInput: mkdir foo\nOutput: Creates directory 'foo'"
},
"run_in_background": {
"type": "boolean",
"description": "Set to true to run this command in the background. Use BashOutput to read the output later."
}
},
"required": [
"command"
],
"additionalProperties": false,
"$schema": "http://json-schema.org/draft-07/schema#"
}
},
{
"name": "Glob",
"description": "- Fast file pattern matching tool that works with any codebase size\n- Supports glob patterns like \"**/*.js\" or \"src/**/*.ts\"\n- Returns matching file paths sorted by modification time\n- Use this tool when you need to find files by name patterns\n- When you are doing an open ended search that may require multiple rounds of globbing and grepping, use the Agent tool instead\n- You have the capability to call multiple tools in a single response. It is always better to speculatively perform multiple searches as a batch that are potentially useful.",
"input_schema": {
"type": "object",
"properties": {
"pattern": {
"type": "string",
"description": "The glob pattern to match files against"
},
"path": {
"type": "string",
"description": "The directory to search in. If not specified, the current working directory will be used. IMPORTANT: Omit this field to use the default directory. DO NOT enter \"undefined\" or \"null\" - simply omit it for the default behavior. Must be a valid directory path if provided."
}
},
"required": [
"pattern"
],
"additionalProperties": false,
"$schema": "http://json-schema.org/draft-07/schema#"
}
},
{
"name": "Grep",
"description": "A powerful search tool built on ripgrep\n\n Usage:\n - ALWAYS use Grep for search tasks. NEVER invoke `grep` or `rg` as a Bash command. The Grep tool has been optimized for correct permissions and access.\n - Supports full regex syntax (e.g., \"log.*Error\", \"function\\s+\\w+\")\n - Filter files with glob parameter (e.g., \"*.js\", \"**/*.tsx\") or type parameter (e.g., \"js\", \"py\", \"rust\")\n - Output modes: \"content\" shows matching lines, \"files_with_matches\" shows only file paths (default), \"count\" shows match counts\n - Use Task tool for open-ended searches requiring multiple rounds\n - Pattern syntax: Uses ripgrep (not grep) - literal braces need escaping (use `interface\\{\\}` to find `interface{}` in Go code)\n - Multiline matching: By default patterns match within single lines only. For cross-line patterns like `struct \\{[\\s\\S]*?field`, use `multiline: true`\n",
"input_schema": {
"type": "object",
"properties": {
"pattern": {
"type": "string",
"description": "The regular expression pattern to search for in file contents"
},
"path": {
"type": "string",
"description": "File or directory to search in (rg PATH). Defaults to current working directory."
},
"glob": {
"type": "string",
"description": "Glob pattern to filter files (e.g. \"*.js\", \"*.{ts,tsx}\") - maps to rg --glob"
},
"output_mode": {
"type": "string",
"enum": [
"content",
"files_with_matches",
"count"
],
"description": "Output mode: \"content\" shows matching lines (supports -A/-B/-C context, -n line numbers, head_limit), \"files_with_matches\" shows file paths (supports head_limit), \"count\" shows match counts (supports head_limit). Defaults to \"files_with_matches\"."
},
"-B": {
"type": "number",
"description": "Number of lines to show before each match (rg -B). Requires output_mode: \"content\", ignored otherwise."
},
"-A": {
"type": "number",
"description": "Number of lines to show after each match (rg -A). Requires output_mode: \"content\", ignored otherwise."
},
"-C": {
"type": "number",
"description": "Number of lines to show before and after each match (rg -C). Requires output_mode: \"content\", ignored otherwise."
},
"-n": {
"type": "boolean",
"description": "Show line numbers in output (rg -n). Requires output_mode: \"content\", ignored otherwise."
},
"-i": {
"type": "boolean",
"description": "Case insensitive search (rg -i)"
},
"type": {
"type": "string",
"description": "File type to search (rg --type). Common types: js, py, rust, go, java, etc. More efficient than include for standard file types."
},
"head_limit": {
"type": "number",
"description": "Limit output to first N lines/entries, equivalent to \"| head -N\". Works across all output modes: content (limits output lines), files_with_matches (limits file paths), count (limits count entries). When unspecified, shows all results from ripgrep."
},
"multiline": {
"type": "boolean",
"description": "Enable multiline mode where . matches newlines and patterns can span lines (rg -U --multiline-dotall). Default: false."
}
},
"required": [
"pattern"
],
"additionalProperties": false,
"$schema": "http://json-schema.org/draft-07/schema#"
}
},
{
"name": "LS",
"description": "Lists files and directories in a given path. The path parameter must be an absolute path, not a relative path. You can optionally provide an array of glob patterns to ignore with the ignore parameter. You should generally prefer the Glob and Grep tools, if you know which directories to search.",
"input_schema": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "The absolute path to the directory to list (must be absolute, not relative)"
},
"ignore": {
"type": "array",
"items": {
"type": "string"
},
"description": "List of glob patterns to ignore"
}
},
"required": [
"path"
],
"additionalProperties": false,
"$schema": "http://json-schema.org/draft-07/schema#"
}
},
{
"name": "ExitPlanMode",
"description": "Use this tool when you are in plan mode and have finished presenting your plan and are ready to code. This will prompt the user to exit plan mode. \nIMPORTANT: Only use this tool when the task requires planning the implementation steps of a task that requires writing code. For research tasks where you're gathering information, searching files, reading files or in general trying to understand the codebase - do NOT use this tool.\n\nEg. \n1. Initial task: \"Search for and understand the implementation of vim mode in the codebase\" - Do not use the exit plan mode tool because you are not planning the implementation steps of a task.\n2. Initial task: \"Help me implement yank mode for vim\" - Use the exit plan mode tool after you have finished planning the implementation steps of the task.\n",
"input_schema": {
"type": "object",
"properties": {
"plan": {
"type": "string",
"description": "The plan you came up with, that you want to run by the user for approval. Supports markdown. The plan should be pretty concise."
}
},
"required": [
"plan"
],
"additionalProperties": false,
"$schema": "http://json-schema.org/draft-07/schema#"
}
},
{
"name": "Read",
"description": "Reads a file from the local filesystem. You can access any file directly by using this tool.\nAssume this tool is able to read all files on the machine. If the User provides a path to a file assume that path is valid. It is okay to read a file that does not exist; an error will be returned.\n\nUsage:\n- The file_path parameter must be an absolute path, not a relative path\n- By default, it reads up to 2000 lines starting from the beginning of the file\n- You can optionally specify a line offset and limit (especially handy for long files), but it's recommended to read the whole file by not providing these parameters\n- Any lines longer than 2000 characters will be truncated\n- Results are returned using cat -n format, with line numbers starting at 1\n- This tool allows Claude Code to read images (eg PNG, JPG, etc). When reading an image file the contents are presented visually as Claude Code is a multimodal LLM.\n- This tool can read PDF files (.pdf). PDFs are processed page by page, extracting both text and visual content for analysis.\n- This tool can read Jupyter notebooks (.ipynb files) and returns all cells with their outputs, combining code, text, and visualizations.\n- You have the capability to call multiple tools in a single response. It is always better to speculatively read multiple files as a batch that are potentially useful. \n- You will regularly be asked to read screenshots. If the user provides a path to a screenshot ALWAYS use this tool to view the file at the path. This tool will work with all temporary file paths like /var/folders/123/abc/T/TemporaryItems/NSIRD_screencaptureui_ZfB1tD/Screenshot.png\n- If you read a file that exists but has empty contents you will receive a system reminder warning in place of file contents.",
"input_schema": {
"type": "object",
"properties": {
"file_path": {
"type": "string",
"description": "The absolute path to the file to read"
},
"offset": {
"type": "number",
"description": "The line number to start reading from. Only provide if the file is too large to read at once"
},
"limit": {
"type": "number",
"description": "The number of lines to read. Only provide if the file is too large to read at once."
}
},
"required": [
"file_path"
],
"additionalProperties": false,
"$schema": "http://json-schema.org/draft-07/schema#"
}
},
{
"name": "Edit",
"description": "Performs exact string replacements in files. \n\nUsage:\n- You must use your `Read` tool at least once in the conversation before editing. This tool will error if you attempt an edit without reading the file. \n- When editing text from Read tool output, ensure you preserve the exact indentation (tabs/spaces) as it appears AFTER the line number prefix. The line number prefix format is: spaces + line number + tab. Everything after that tab is the actual file content to match. Never include any part of the line number prefix in the old_string or new_string.\n- ALWAYS prefer editing existing files in the codebase. NEVER write new files unless explicitly required.\n- Only use emojis if the user explicitly requests it. Avoid adding emojis to files unless asked.\n- The edit will FAIL if `old_string` is not unique in the file. Either provide a larger string with more surrounding context to make it unique or use `replace_all` to change every instance of `old_string`. \n- Use `replace_all` for replacing and renaming strings across the file. This parameter is useful if you want to rename a variable for instance.",
"input_schema": {
"type": "object",
"properties": {
"file_path": {
"type": "string",
"description": "The absolute path to the file to modify"
},
"old_string": {
"type": "string",
"description": "The text to replace"
},
"new_string": {
"type": "string",
"description": "The text to replace it with (must be different from old_string)"
},
"replace_all": {
"type": "boolean",
"default": false,
"description": "Replace all occurences of old_string (default false)"
}
},
"required": [
"file_path",
"old_string",
"new_string"
],
"additionalProperties": false,
"$schema": "http://json-schema.org/draft-07/schema#"
}
},
{
"name": "MultiEdit",
"description": "This is a tool for making multiple edits to a single file in one operation. It is built on top of the Edit tool and allows you to perform multiple find-and-replace operations efficiently. Prefer this tool over the Edit tool when you need to make multiple edits to the same file.\n\nBefore using this tool:\n\n1. Use the Read tool to understand the file's contents and context\n2. Verify the directory path is correct\n\nTo make multiple file edits, provide the following:\n1. file_path: The absolute path to the file to modify (must be absolute, not relative)\n2. edits: An array of edit operations to perform, where each edit contains:\n - old_string: The text to replace (must match the file contents exactly, including all whitespace and indentation)\n - new_string: The edited text to replace the old_string\n - replace_all: Replace all occurences of old_string. This parameter is optional and defaults to false.\n\nIMPORTANT:\n- All edits are applied in sequence, in the order they are provided\n- Each edit operates on the result of the previous edit\n- All edits must be valid for the operation to succeed - if any edit fails, none will be applied\n- This tool is ideal when you need to make several changes to different parts of the same file\n- For Jupyter notebooks (.ipynb files), use the NotebookEdit instead\n\nCRITICAL REQUIREMENTS:\n1. All edits follow the same requirements as the single Edit tool\n2. The edits are atomic - either all succeed or none are applied\n3. Plan your edits carefully to avoid conflicts between sequential operations\n\nWARNING:\n- The tool will fail if edits.old_string doesn't match the file contents exactly (including whitespace)\n- The tool will fail if edits.old_string and edits.new_string are the same\n- Since edits are applied in sequence, ensure that earlier edits don't affect the text that later edits are trying to find\n\nWhen making edits:\n- Ensure all edits result in idiomatic, correct code\n- Do not leave the code in a broken state\n- Always use absolute file paths (starting with /)\n- Only use emojis if the user explicitly requests it. Avoid adding emojis to files unless asked.\n- Use replace_all for replacing and renaming strings across the file. This parameter is useful if you want to rename a variable for instance.\n\nIf you want to create a new file, use:\n- A new file path, including dir name if needed\n- First edit: empty old_string and the new file's contents as new_string\n- Subsequent edits: normal edit operations on the created content",
"input_schema": {
"type": "object",
"properties": {
"file_path": {
"type": "string",
"description": "The absolute path to the file to modify"
},
"edits": {
"type": "array",
"items": {
"type": "object",
"properties": {
"old_string": {
"type": "string",
"description": "The text to replace"
},
"new_string": {
"type": "string",
"description": "The text to replace it with"
},
"replace_all": {
"type": "boolean",
"default": false,
"description": "Replace all occurences of old_string (default false)."
}
},
"required": [
"old_string",
"new_string"
],
"additionalProperties": false
},
"minItems": 1,
"description": "Array of edit operations to perform sequentially on the file"
}
},
"required": [
"file_path",
"edits"
],
"additionalProperties": false,
"$schema": "http://json-schema.org/draft-07/schema#"
}
},
{
"name": "Write",
"description": "Writes a file to the local filesystem.\n\nUsage:\n- This tool will overwrite the existing file if there is one at the provided path.\n- If this is an existing file, you MUST use the Read tool first to read the file's contents. This tool will fail if you did not read the file first.\n- ALWAYS prefer editing existing files in the codebase. NEVER write new files unless explicitly required.\n- NEVER proactively create documentation files (*.md) or README files. Only create documentation files if explicitly requested by the User.\n- Only use emojis if the user explicitly requests it. Avoid writing emojis to files unless asked.",
"input_schema": {
"type": "object",
"properties": {
"file_path": {
"type": "string",
"description": "The absolute path to the file to write (must be absolute, not relative)"
},
"content": {
"type": "string",
"description": "The content to write to the file"
}
},
"required": [
"file_path",
"content"
],
"additionalProperties": false,
"$schema": "http://json-schema.org/draft-07/schema#"
}
},
{
"name": "NotebookEdit",
"description": "Completely replaces the contents of a specific cell in a Jupyter notebook (.ipynb file) with new source. Jupyter notebooks are interactive documents that combine code, text, and visualizations, commonly used for data analysis and scientific computing. The notebook_path parameter must be an absolute path, not a relative path. The cell_number is 0-indexed. Use edit_mode=insert to add a new cell at the index specified by cell_number. Use edit_mode=delete to delete the cell at the index specified by cell_number.",
"input_schema": {
"type": "object",
"properties": {
"notebook_path": {
"type": "string",
"description": "The absolute path to the Jupyter notebook file to edit (must be absolute, not relative)"
},
"cell_id": {
"type": "string",
"description": "The ID of the cell to edit. When inserting a new cell, the new cell will be inserted after the cell with this ID, or at the beginning if not specified."
},
"new_source": {
"type": "string",
"description": "The new source for the cell"
},
"cell_type": {
"type": "string",
"enum": [
"code",
"markdown"
],
"description": "The type of the cell (code or markdown). If not specified, it defaults to the current cell type. If using edit_mode=insert, this is required."
},
"edit_mode": {
"type": "string",
"enum": [
"replace",
"insert",
"delete"
],
"description": "The type of edit to make (replace, insert, delete). Defaults to replace."
}
},
"required": [
"notebook_path",
"new_source"
],
"additionalProperties": false,
"$schema": "http://json-schema.org/draft-07/schema#"
}
},
{
"name": "WebFetch",
"description": "\n- Fetches content from a specified URL and processes it using an AI model\n- Takes a URL and a prompt as input\n- Fetches the URL content, converts HTML to markdown\n- Processes the content with the prompt using a small, fast model\n- Returns the model's response about the content\n- Use this tool when you need to retrieve and analyze web content\n\nUsage notes:\n - IMPORTANT: If an MCP-provided web fetch tool is available, prefer using that tool instead of this one, as it may have fewer restrictions. All MCP-provided tools start with \"mcp__\".\n - The URL must be a fully-formed valid URL\n - HTTP URLs will be automatically upgraded to HTTPS\n - The prompt should describe what information you want to extract from the page\n - This tool is read-only and does not modify any files\n - Results may be summarized if the content is very large\n - Includes a self-cleaning 15-minute cache for faster responses when repeatedly accessing the same URL\n - When a URL redirects to a different host, the tool will inform you and provide the redirect URL in a special format. You should then make a new WebFetch request with the redirect URL to fetch the content.\n",
"input_schema": {
"type": "object",
"properties": {
"url": {
"type": "string",
"format": "uri",
"description": "The URL to fetch content from"
},
"prompt": {
"type": "string",
"description": "The prompt to run on the fetched content"
}
},
"required": [
"url",
"prompt"
],
"additionalProperties": false,
"$schema": "http://json-schema.org/draft-07/schema#"
}
},
{
"name": "TodoWrite",
"description": "Use this tool to create and manage a structured task list for your current coding session. This helps you track progress, organize complex tasks, and demonstrate thoroughness to the user.\nIt also helps the user understand the progress of the task and overall progress of their requests.\n\n## When to Use This Tool\nUse this tool proactively in these scenarios:\n\n1. Complex multi-step tasks - When a task requires 3 or more distinct steps or actions\n2. Non-trivial and complex tasks - Tasks that require careful planning or multiple operations\n3. User explicitly requests todo list - When the user directly asks you to use the todo list\n4. User provides multiple tasks - When users provide a list of things to be done (numbered or comma-separated)\n5. After receiving new instructions - Immediately capture user requirements as todos\n6. When you start working on a task - Mark it as in_progress BEFORE beginning work. Ideally you should only have one todo as in_progress at a time\n7. After completing a task - Mark it as completed and add any new follow-up tasks discovered during implementation\n\n## When NOT to Use This Tool\n\nSkip using this tool when:\n1. There is only a single, straightforward task\n2. The task is trivial and tracking it provides no organizational benefit\n3. The task can be completed in less than 3 trivial steps\n4. The task is purely conversational or informational\n\nNOTE that you should not use this tool if there is only one trivial task to do. In this case you are better off just doing the task directly.\n\n## Examples of When to Use the Todo List\n\n<example>\nUser: I want to add a dark mode toggle to the application settings. Make sure you run the tests and build when you're done!\nAssistant: I'll help add a dark mode toggle to your application settings. Let me create a todo list to track this implementation.\n*Creates todo list with the following items:*\n1. Create dark mode toggle component in Settings page\n2. Add dark mode state management (context/store)\n3. Implement CSS-in-JS styles for dark theme\n4. Update existing components to support theme switching\n5. Run tests and build process, addressing any failures or errors that occur\n*Begins working on the first task*\n\n<reasoning>\nThe assistant used the todo list because:\n1. Adding dark mode is a multi-step feature requiring UI, state management, and styling changes\n2. The user explicitly requested tests and build be run afterward\n3. The assistant inferred that tests and build need to pass by adding \"Ensure tests and build succeed\" as the final task\n</reasoning>\n</example>\n\n<example>\nUser: Help me rename the function getCwd to getCurrentWorkingDirectory across my project\nAssistant: Let me first search through your codebase to find all occurrences of 'getCwd'.\n*Uses grep or search tools to locate all instances of getCwd in the codebase*\nAssistant: I've found 15 instances of 'getCwd' across 8 different files. Let me create a todo list to track these changes.\n*Creates todo list with specific items for each file that needs updating*\n\n<reasoning>\nThe assistant used the todo list because:\n1. First, the assistant searched to understand the scope of the task\n2. Upon finding multiple occurrences across different files, it determined this was a complex task with multiple steps\n3. The todo list helps ensure every instance is tracked and updated systematically\n4. This approach prevents missing any occurrences and maintains code consistency\n</reasoning>\n</example>\n\n\n<example>\nUser: I need to implement these features for my e-commerce site: user registration, product catalog, shopping cart, and checkout flow.\nAssistant: I'll help implement these features. First, let's add all the features to the todo list.\n*Creates a todo list breaking down each feature into specific tasks based on the project architecture*\nAssistant: Let's start with user registration. This will involve creating database models, API endpoints, and frontend forms that integrate with your existing authentication system.\n\n<reasoning>\nThe assistant used the todo list because:\n1. The user provided multiple complex features to implement in a comma separated list\n2. The todo list helps organize these large features into manageable tasks\n3. This approach allows for tracking progress across the entire implementation\n</reasoning>\n</example>\n\n<example>\nUser: Can you help optimize my React application? It's rendering slowly and has performance issues.</user>\nAssistant: I'll help optimize your React application. First, let me examine your codebase to identify potential performance bottlenecks.\n*Reviews component structure, render patterns, state management, and data fetching*\nAssistant: After analyzing your codebase, I've identified several performance issues. Let me create a todo list to track our optimization efforts.\n*Creates todo list with items like: 1) Implement memoization for expensive calculations in ProductList, 2) Add virtualization for long lists in Dashboard, 3) Optimize image loading in Gallery component, 4) Fix state update loops in ShoppingCart, 5) Review bundle size and implement code splitting*\nLet's start by implementing memoization for the expensive calculations in your ProductList component.</assistant>\n\n<reasoning>\nThe assistant used the todo list because:\n1. First, the assistant examined the codebase to identify specific performance issues\n2. Based on this analysis, it identified multiple optimization opportunities\n3. Performance optimization is a non-trivial task requiring multiple steps\n4. The todo list helps methodically track improvements across different components\n5. This systematic approach ensures all performance bottlenecks are addressed\n</reasoning>\n</example>\n\n## Examples of When NOT to Use the Todo List\n\n<example>\nUser: How do I print 'Hello World' in Python?\nAssistant: In Python, you can print \"Hello World\" with this simple code:\n\npython\nprint(\"Hello World\")\n\nThis will output the text \"Hello World\" to the console when executed.</assistant>\n\n<reasoning>\nThe assistant did not use the todo list because this is a single, trivial task that can be completed in one step. There's no need to track multiple tasks or steps for such a straightforward request.\n</reasoning>\n</example>\n\n<example>\nUser: What does the git status command do?\nAssistant: The git status command shows the current state of your working directory and staging area. It displays which changes have been staged, which haven't, and which files aren't being tracked by Git. This is a fundamental command for understanding the current state of your repository before making commits.\n\n<reasoning>\nThe assistant did not use the todo list because this is an informational request with no actual coding task to complete. The user is simply asking for an explanation, not for the assistant to perform multiple steps or tasks.\n</reasoning>\n</example>\n\n<example>\nUser: Can you add a comment to the calculateTotal function to explain what it does?\nAssistant: Sure, let me add a comment to the calculateTotal function to explain what it does.\n* Uses the Edit tool to add a comment to the calculateTotal function *\n\n<reasoning>\nThe assistant did not use the todo list because this is a single, straightforward task confined to one location in the code. Adding a comment doesn't require tracking multiple steps or systematic organization.\n</reasoning>\n</example>\n\n<example>\nUser: Run npm install for me and tell me what happens.\nAssistant: I'll run the npm install command for you.\n\n*Executes: npm install*\n\nThe command completed successfully. Here's the output:\n[Output of npm install command]\n\nAll dependencies have been installed according to your package.json file.\n\n<reasoning>\nThe assistant did not use the todo list because this is a single command execution with immediate results. There are no multiple steps to track or organize, making the todo list unnecessary for this straightforward task.\n</reasoning>\n</example>\n\n## Task States and Management\n\n1. **Task States**: Use these states to track progress:\n - pending: Task not yet started\n - in_progress: Currently working on (limit to ONE task at a time)\n - completed: Task finished successfully\n\n2. **Task Management**:\n - Update task status in real-time as you work\n - Mark tasks complete IMMEDIATELY after finishing (don't batch completions)\n - Only have ONE task in_progress at any time\n - Complete current tasks before starting new ones\n - Remove tasks that are no longer relevant from the list entirely\n\n3. **Task Completion Requirements**:\n - ONLY mark a task as completed when you have FULLY accomplished it\n - If you encounter errors, blockers, or cannot finish, keep the task as in_progress\n - When blocked, create a new task describing what needs to be resolved\n - Never mark a task as completed if:\n - Tests are failing\n - Implementation is partial\n - You encountered unresolved errors\n - You couldn't find necessary files or dependencies\n\n4. **Task Breakdown**:\n - Create specific, actionable items\n - Break complex tasks into smaller, manageable steps\n - Use clear, descriptive task names\n\nWhen in doubt, use this tool. Being proactive with task management demonstrates attentiveness and ensures you complete all requirements successfully.\n",
"input_schema": {
"type": "object",
"properties": {
"todos": {
"type": "array",
"items": {
"type": "object",
"properties": {
"content": {
"type": "string",
"minLength": 1
},
"status": {
"type": "string",
"enum": [
"pending",
"in_progress",
"completed"
]
},
"id": {
"type": "string"
}
},
"required": [
"content",
"status",
"id"
],
"additionalProperties": false
},
"description": "The updated todo list"
}
},
"required": [
"todos"
],
"additionalProperties": false,
"$schema": "http://json-schema.org/draft-07/schema#"
}
},
{
"name": "WebSearch",
"description": "\n- Allows Claude to search the web and use the results to inform responses\n- Provides up-to-date information for current events and recent data\n- Returns search result information formatted as search result blocks\n- Use this tool for accessing information beyond Claude's knowledge cutoff\n- Searches are performed automatically within a single API call\n\nUsage notes:\n - Domain filtering is supported to include or block specific websites\n - Web search is only available in the US\n - Account for \"Today's date\" in <env>. For example, if <env> says \"Today's date: 2025-07-01\", and the user wants the latest docs, do not use 2024 in the search query. Use 2025.\n",
"input_schema": {
"type": "object",
"properties": {
"query": {
"type": "string",
"minLength": 2,
"description": "The search query to use"
},
"allowed_domains": {
"type": "array",
"items": {
"type": "string"
},
"description": "Only include search results from these domains"
},
"blocked_domains": {
"type": "array",
"items": {
"type": "string"
},
"description": "Never include search results from these domains"
}
},
"required": [
"query"
],
"additionalProperties": false,
"$schema": "http://json-schema.org/draft-07/schema#"
}
},
{
"name": "BashOutput",
"description": "\n- Retrieves output from a running or completed background bash shell\n- Takes a shell_id parameter identifying the shell\n- Always returns only new output since the last check\n- Returns stdout and stderr output along with shell status\n- Supports optional regex filtering to show only lines matching a pattern\n- Use this tool when you need to monitor or check the output of a long-running shell\n- Shell IDs can be found using the /bashes command\n",
"input_schema": {
"type": "object",
"properties": {
"bash_id": {
"type": "string",
"description": "The ID of the background shell to retrieve output from"
},
"filter": {
"type": "string",
"description": "Optional regular expression to filter the output lines. Only lines matching this regex will be included in the result. Any lines that do not match will no longer be available to read."
}
},
"required": [
"bash_id"
],
"additionalProperties": false,
"$schema": "http://json-schema.org/draft-07/schema#"
}
},
{
"name": "KillBash",
"description": "\n- Kills a running background bash shell by its ID\n- Takes a shell_id parameter identifying the shell to kill\n- Returns a success or failure status \n- Use this tool when you need to terminate a long-running shell\n- Shell IDs can be found using the /bashes command\n",
"input_schema": {
"type": "object",
"properties": {
"shell_id": {
"type": "string",
"description": "The ID of the background shell to kill"
}
},
"required": [
"shell_id"
],
"additionalProperties": false,
"$schema": "http://json-schema.org/draft-07/schema#"
}
}
]
}