clean up

Signed-off-by: Jintao Zhang <zhangjintao9020@gmail.com>

.github/ISSUE_TEMPLATE/bug_report.md

@@ -0,0 +1,34 @@
+---
+name: Bug Report
+about: Create a report to help us improve
+title: '[BUG] '
+labels: bug
+assignees: ''
+---
+
+## Bug Description
+A clear and concise description of what the bug is.
+
+## To Reproduce
+Steps to reproduce the behavior:
+1. Run command '...'
+2. See error
+
+## Expected Behavior
+A clear and concise description of what you expected to happen.
+
+## Actual Behavior
+What actually happened.
+
+## Environment
+- OS: [e.g. Ubuntu 22.04]
+- Python Version: [e.g. 3.11.5]
+- AMCP Version: [e.g. 0.1.0]
+
+## Additional Context
+Add any other context about the problem here.
+
+## Logs
+```
+Paste relevant logs here
+```

.github/ISSUE_TEMPLATE/feature_request.md

@@ -0,0 +1,22 @@
+---
+name: Feature Request
+about: Suggest an idea for this project
+title: '[FEATURE] '
+labels: enhancement
+assignees: ''
+---
+
+## Feature Description
+A clear and concise description of what you want to happen.
+
+## Use Case
+Describe the use case or problem this feature would solve.
+
+## Proposed Solution
+Describe how you envision this feature working.
+
+## Alternatives Considered
+A clear and concise description of any alternative solutions or features you've considered.
+
+## Additional Context
+Add any other context or screenshots about the feature request here.

.github/PULL_REQUEST_TEMPLATE.md

@@ -0,0 +1,33 @@
+# Pull Request
+
+## Description
+<!-- Describe your changes in detail -->
+
+## Type of Change
+- [ ] Bug fix (non-breaking change which fixes an issue)
+- [ ] New feature (non-breaking change which adds functionality)
+- [ ] Breaking change (fix or feature that would cause existing functionality to not work as expected)
+- [ ] Documentation update
+
+## Checklist
+- [ ] My code follows the style guidelines of this project
+- [ ] I have performed a self-review of my own code
+- [ ] I have commented my code, particularly in hard-to-understand areas
+- [ ] I have made corresponding changes to the documentation
+- [ ] My changes generate no new warnings
+- [ ] I have added tests that prove my fix is effective or that my feature works
+- [ ] New and existing unit tests pass locally with my changes
+- [ ] Any dependent changes have been merged and published
+
+## Testing
+<!-- Describe the tests you ran to verify your changes -->
+
+```bash
+# Example test commands
+make test
+make lint
+```
+
+## Related Issues
+<!-- Link to related issues -->
+Closes #

.github/workflows/ci.yml

@@ -0,0 +1,44 @@
+name: CI
+
+on:
+  push:
+    branches: [main]
+  pull_request:
+    branches: [main]
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: ["3.11", "3.12", "3.13"]
+
+    steps:
+    - uses: actions/checkout@v4
+    
+    - name: Set up Python ${{ matrix.python-version }}
+      uses: actions/setup-python@v5
+      with:
+        python-version: ${{ matrix.python-version }}
+    
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        pip install -e ".[dev]"
+    
+    - name: Lint with ruff
+      run: ruff check src/
+    
+    - name: Type check with mypy
+      run: mypy src/amcp --ignore-missing-imports
+      continue-on-error: true
+    
+    - name: Test with pytest
+      run: pytest --cov --cov-report=xml
+    
+    - name: Upload coverage
+      uses: codecov/codecov-action@v4
+      if: matrix.python-version == '3.11'
+      with:
+        file: ./coverage.xml
+        fail_ci_if_error: false

.pre-commit-config.yaml

@@ -0,0 +1,15 @@
+repos:
+  - repo: https://github.com/astral-sh/ruff-pre-commit
+    rev: v0.3.0
+    hooks:
+      - id: ruff
+        args: [--fix]
+      - id: ruff-format
+
+  - repo: https://github.com/pre-commit/pre-commit-hooks
+    rev: v4.5.0
+    hooks:
+      - id: trailing-whitespace
+      - id: end-of-file-fixer
+      - id: check-yaml
+      - id: check-added-large-files

.ruff.toml

@@ -0,0 +1,9 @@
+# Ruff configuration
+line-length = 120
+
+[lint]
+select = ["E", "F", "UP", "B", "SIM", "I"]
+ignore = ["E501"]  # Ignore line length for now
+
+[lint.per-file-ignores]
+"tests/*" = ["F401", "F811"]

CONTRIBUTING.md

@@ -0,0 +1,64 @@
+# Contributing to AMCP
+
+## Development Setup
+
+```bash
+# Clone the repository
+git clone <repo-url>
+cd amcp
+
+# Install in editable mode with dev dependencies
+pip install -e ".[dev]"
+
+# Or using uv
+uv pip install -e ".[dev]"
+```
+
+## Running Tests
+
+```bash
+# Run all tests
+make test
+
+# Run with coverage
+make test-cov
+
+# Run specific test
+pytest tests/test_config.py
+```
+
+## Code Quality
+
+```bash
+# Lint code
+make lint
+
+# Format code
+make format
+
+# Type check
+make type-check
+```
+
+## Pre-commit Hooks
+
+```bash
+pip install pre-commit
+pre-commit install
+```
+
+## Project Structure
+
+```
+amcp/
+├── src/amcp/          # Main package
+│   ├── __init__.py
+│   ├── cli.py         # CLI entry point
+│   ├── agent.py       # Agent logic
+│   ├── tools.py       # Built-in tools
+│   ├── config.py      # Configuration
+│   └── mcp_client.py  # MCP integration
+├── tests/             # Test suite
+├── .github/workflows/ # CI/CD
+└── pyproject.toml     # Project metadata
+```

Dockerfile

@@ -20,8 +20,5 @@ RUN mkdir -p /root/.config/amcp
 # Set environment variables
 ENV PYTHONUNBUFFERED=1
 
-# Expose port for Gradio
-EXPOSE 7860
-
-# Run Gradio app
-CMD ["python", "app.py"]
+# Default command
+CMD ["amcp", "--help"]

Makefile

@@ -0,0 +1,23 @@
+.PHONY: install test lint format clean
+
+install:
+	pip install -e ".[dev]"
+
+test:
+	pytest
+
+test-cov:
+	pytest --cov --cov-report=html --cov-report=term
+
+lint:
+	ruff check src/ tests/
+
+format:
+	ruff format src/ tests/
+
+type-check:
+	mypy src/amcp --ignore-missing-imports
+
+clean:
+	rm -rf build/ dist/ *.egg-info .pytest_cache .coverage htmlcov/
+	find . -type d -name __pycache__ -exec rm -rf {} +

README.md

@@ -1,21 +1,9 @@
----
-title: AMCP
-emoji: 🤖
-colorFrom: blue
-colorTo: yellow
-sdk: gradio
-sdk_version: 6.0.1
-app_file: app.py
-python_version: "3.11"
-pinned: false
-license: apache-2.0
-tags:
-- building-mcp-track-creative
-- mcp-in-action-track-consumer
-- mcp-in-action-track-creative
----
-
 # AMCP
+
+[![CI](https://github.com/tao12345666333/amcp/workflows/CI/badge.svg)](https://github.com/tao12345666333/amcp/actions)
+[![Python 3.11+](https://img.shields.io/badge/python-3.11+-blue.svg)](https://www.python.org/downloads/)
+[![License: Apache-2.0](https://img.shields.io/badge/License-Apache%202.0-blue.svg)](https://opensource.org/licenses/Apache-2.0)
+
 tags:
 - building-mcp-track-creative
 - mcp-in-action-track-consumer
@@ -107,10 +95,54 @@ write_tool_enabled = true  # Enable/disable built-in write_file tool
 edit_tool_enabled = true   # Enable/disable built-in edit_file tool
 ```
 
+## Development
+
+### Setup Development Environment
+
+```bash
+# Clone the repository
+git clone <repo-url>
+cd AMCP
+
+# Install with development dependencies
+pip install -e ".[dev]"
+
+# Install pre-commit hooks
+pre-commit install
+```
+
+### Running Tests
+
+```bash
+# Run all tests
+make test
+
+# Run with coverage
+make test-cov
+
+# Run specific test
+pytest tests/test_tools.py -v
+```
+
+### Code Quality
+
+```bash
+# Lint code
+make lint
+
+# Format code
+make format
+
+# Type check
+make type-check
+```
+
+See [CONTRIBUTING.md](CONTRIBUTING.md) for detailed development guidelines.
+
 ## Notes
 - `rg` (ripgrep) must be installed and on PATH for the grep tool.
 - MCP servers must be installed separately and runnable (stdio transport).
 
 ## License
 
-MIT
+Apache-2.0

app.py

@@ -1,37 +0,0 @@
-import gradio as gr
-import asyncio
-from src.amcp.agent import Agent
-from src.amcp.agent_spec import get_default_agent_spec
-
-async def chat(message, history):
-    """Process chat message with AMCP agent."""
-    agent = Agent(agent_spec=get_default_agent_spec())
-    
-    try:
-        response = await agent.run(
-            user_input=message,
-            stream=False,
-            show_progress=False
-        )
-        return response
-    except Exception as e:
-        return f"Error: {str(e)}"
-
-def chat_wrapper(message, history):
-    """Sync wrapper for async chat function."""
-    return asyncio.run(chat(message, history))
-
-# Create Gradio interface
-demo = gr.ChatInterface(
-    fn=chat_wrapper,
-    title="AMCP - Agent CLI",
-    description="A Lego-style coding agent with built-in tools (read_file, grep, bash, think)",
-    examples=[
-        "List files in the current directory",
-        "Search for 'def' in Python files",
-        "What tools do you have available?"
-    ]
-)
-
-if __name__ == "__main__":
-    demo.launch()

docs/PROJECT_STRUCTURE.md

@@ -0,0 +1,90 @@
+# Project Structure
+
+## Overview
+
+AMCP follows Python best practices with a clear separation of concerns:
+
+```
+AMCP/
+├── src/amcp/              # Main package source code
+│   ├── __init__.py        # Package initialization
+│   ├── __main__.py        # Entry point for python -m amcp
+│   ├── cli.py             # CLI interface (Typer)
+│   ├── agent.py           # Agent orchestration logic
+│   ├── agent_spec.py      # Agent specification handling
+│   ├── tools.py           # Built-in tools (read, grep, bash, etc.)
+│   ├── config.py          # Configuration management
+│   ├── mcp_client.py      # MCP server integration
+│   ├── chat.py            # Chat/LLM interaction
+│   └── readfile.py        # File reading utilities
+│
+├── tests/                 # Test suite
+│   ├── __init__.py
+│   ├── conftest.py        # Pytest fixtures
+│   ├── test_agent_spec.py
+│   ├── test_config.py
+│   └── test_tools.py
+│
+├── .github/
+│   └── workflows/
+│       └── ci.yml         # GitHub Actions CI/CD
+│
+├── docs/                  # Documentation
+│   └── PROJECT_STRUCTURE.md
+│
+├── pyproject.toml         # Project metadata & dependencies
+├── pytest.ini             # Pytest configuration
+├── Makefile               # Common development tasks
+├── .pre-commit-config.yaml # Pre-commit hooks
+├── .ruff.toml             # Ruff linter configuration
+├── .gitignore
+├── README.md
+├── CONTRIBUTING.md
+├── CHANGELOG.md
+└── Dockerfile
+```
+
+## Key Design Decisions
+
+### 1. Source Layout (`src/` layout)
+- Prevents accidental imports of uninstalled code
+- Clear separation between source and tests
+- Recommended by PyPA
+
+### 2. Testing
+- Uses pytest for testing framework
+- Fixtures in `conftest.py` for reusability
+- Coverage reporting with pytest-cov
+- Target: >80% code coverage
+
+### 3. Code Quality
+- Ruff for linting and formatting
+- Type hints encouraged (mypy for type checking)
+- Pre-commit hooks for automated checks
+
+### 4. CI/CD
+- GitHub Actions for automated testing
+- Matrix testing across Python 3.11, 3.12, 3.13
+- Automated coverage reporting
+
+### 5. Configuration
+- pyproject.toml as single source of truth
+- Tool configurations centralized
+- Optional dependencies for development
+
+## Development Workflow
+
+1. **Setup**: `make install` or `pip install -e ".[dev]"`
+2. **Test**: `make test` or `pytest`
+3. **Lint**: `make lint` or `ruff check src/ tests/`
+4. **Format**: `make format` or `ruff format src/ tests/`
+5. **Coverage**: `make test-cov`
+
+## Module Responsibilities
+
+- **cli.py**: Command-line interface, argument parsing
+- **agent.py**: Core agent logic, tool orchestration
+- **tools.py**: Built-in tool implementations
+- **config.py**: Configuration loading/saving
+- **mcp_client.py**: MCP protocol communication
+- **chat.py**: LLM interaction, streaming responses

docs/QUICK_START.md

@@ -0,0 +1,157 @@
+# Quick Start Guide
+
+## For Contributors
+
+### 1. Clone and Setup
+```bash
+git clone <repo-url>
+cd AMCP
+
+# Install with development dependencies
+pip install -e ".[dev]"
+
+# Or using uv (recommended)
+uv pip install -e ".[dev]"
+```
+
+### 2. Install Pre-commit Hooks
+```bash
+pre-commit install
+```
+
+### 3. Run Tests
+```bash
+# Quick test
+make test
+
+# With coverage report
+make test-cov
+
+# Specific test file
+pytest tests/test_tools.py -v
+```
+
+### 4. Code Quality Checks
+```bash
+# Lint code
+make lint
+
+# Auto-format code
+make format
+
+# Type check
+make type-check
+
+# Run all checks
+make lint && make format && make test
+```
+
+### 5. Development Workflow
+```bash
+# 1. Create a feature branch
+git checkout -b feature/my-feature
+
+# 2. Make changes and test
+make test
+
+# 3. Format and lint
+make format
+make lint
+
+# 4. Commit (pre-commit hooks will run automatically)
+git add .
+git commit -m "feat: add new feature"
+
+# 5. Push and create PR
+git push origin feature/my-feature
+```
+
+## For Users
+
+### Installation
+```bash
+pip install amcp
+```
+
+### Basic Usage
+```bash
+# Interactive mode
+amcp
+
+# Single command
+amcp --once "create a hello.py file"
+
+# List available agents
+amcp --list
+
+# Use specific agent
+amcp --agent path/to/agent.yaml
+```
+
+## Common Tasks
+
+### Adding a New Test
+```python
+# tests/test_myfeature.py
+import pytest
+from amcp.myfeature import MyClass
+
+def test_my_feature():
+    obj = MyClass()
+    result = obj.do_something()
+    assert result == expected_value
+```
+
+### Adding a New Tool
+```python
+# src/amcp/tools.py
+class MyNewTool(BaseTool):
+    @property
+    def name(self) -> str:
+        return "my_tool"
+    
+    @property
+    def description(self) -> str:
+        return "Description of what this tool does"
+    
+    def execute(self, **kwargs) -> ToolResult:
+        # Implementation
+        return ToolResult(success=True, content="result")
+```
+
+### Running CI Locally
+```bash
+# Run the same checks as CI
+make lint
+make type-check
+make test-cov
+```
+
+## Troubleshooting
+
+### Tests Failing
+```bash
+# Run with verbose output
+pytest -vv
+
+# Run specific test
+pytest tests/test_tools.py::test_read_file_tool -v
+
+# Show print statements
+pytest -s
+```
+
+### Import Errors
+```bash
+# Reinstall in editable mode
+pip install -e .
+```
+
+### Pre-commit Issues
+```bash
+# Update hooks
+pre-commit autoupdate
+
+# Run manually
+pre-commit run --all-files
+```

pyproject.toml

@@ -27,7 +27,26 @@ requires = ["hatchling>=1.25.0"]
 build-backend = "hatchling.build"
 
 [tool.ruff]
-line-length = 100
+line-length = 120
 
 [tool.ruff.lint]
 select = ["E", "F", "UP", "B", "SIM", "I"]
+ignore = ["E501"]
+
+[project.optional-dependencies]
+dev = [
+  "pytest>=8.0.0",
+  "pytest-cov>=4.1.0",
+  "ruff>=0.3.0",
+  "mypy>=1.8.0",
+]
+
+[tool.pytest.ini_options]
+testpaths = ["tests"]
+addopts = "-v --cov=src/amcp --cov-report=term-missing"
+
+[tool.mypy]
+python_version = "3.11"
+warn_return_any = true
+warn_unused_configs = true
+disallow_untyped_defs = false

pytest.ini

@@ -0,0 +1,6 @@
+[pytest]
+testpaths = tests
+python_files = test_*.py
+python_classes = Test*
+python_functions = test_*
+addopts = -v --tb=short

requirements.txt

@@ -1,9 +0,0 @@
-gradio>=4.0.0
-openai>=1.0.0
-pydantic>=2.0.0
-pyyaml>=6.0.0
-typer>=0.9.0
-rich>=13.0.0
-mcp>=1.0.0
-tomli>=2.0.0
-tomli-w>=1.0.0

src/amcp/agent.py

@@ -1,32 +1,31 @@
 from __future__ import annotations
 
-import asyncio
 import json
-import os
 from datetime import datetime
 from pathlib import Path
-from typing import Any, Dict, List, Optional, Union
+from typing import Any
 
 import typer
 from rich.console import Console
 from rich.panel import Panel
-from rich.progress import Progress, SpinnerColumn, TextColumn
 from rich.status import Status
 
 from .agent_spec import ResolvedAgentSpec, get_default_agent_spec
-from .chat import _chat_with_tools, _make_client, _resolve_base_url, _resolve_api_key
-from .config import AMCPConfig, load_config
+from .chat import _make_client, _resolve_api_key, _resolve_base_url
+from .config import load_config
 from .mcp_client import call_mcp_tool, list_mcp_tools
-from .tools import ToolRegistry, ToolResult
+from .tools import ToolRegistry
 
 
 class AgentExecutionError(Exception):
     """Raised when agent execution fails."""
+
     pass
 
 
 class MaxStepsReached(Exception):
     """Raised when agent reaches maximum execution steps."""
+
     pass
 
 
@@ -42,21 +41,21 @@ class Agent:
     - Conversation history persistence
     """
 
-    def __init__(self, agent_spec: Optional[ResolvedAgentSpec] = None, session_id: Optional[str] = None):
+    def __init__(self, agent_spec: ResolvedAgentSpec | None = None, session_id: str | None = None):
         self.agent_spec = agent_spec or get_default_agent_spec()
         self.console = Console()
         self.tool_registry = ToolRegistry()
-        self.execution_context: Dict[str, Any] = {}
+        self.execution_context: dict[str, Any] = {}
         self.step_count = 0
-        self.tool_calls_history: List[Dict[str, Any]] = []
+        self.tool_calls_history: list[dict[str, Any]] = []
 
         # Conversation history management
         self.session_id = session_id or self._generate_session_id()
-        self.conversation_history: List[Dict[str, Any]] = []
+        self.conversation_history: list[dict[str, Any]] = []
         self.session_file = Path.home() / ".config" / "amcp" / "sessions" / f"{self.session_id}.json"
 
         # Tool call tracking for per-conversation and per-session limits
-        self.current_conversation_tool_calls: List[Dict[str, Any]] = []
+        self.current_conversation_tool_calls: list[dict[str, Any]] = []
 
         # Load existing conversation history if available
         self._load_conversation_history()
@@ -68,6 +67,7 @@ class Agent:
     def _generate_session_id(self) -> str:
         """Generate a unique session ID."""
         import uuid
+
         return str(uuid.uuid4())[:8]
 
     def _ensure_sessions_dir(self) -> None:
@@ -80,12 +80,14 @@ class Agent:
         try:
             if self.session_file.exists():
                 self._ensure_sessions_dir()
-                with open(self.session_file, 'r', encoding='utf-8') as f:
+                with open(self.session_file, encoding="utf-8") as f:
                     data = json.load(f)
-                    self.conversation_history = data.get('conversation_history', [])
-                    self.tool_calls_history = data.get('tool_calls_history', [])
-                    self.current_conversation_tool_calls = data.get('current_conversation_tool_calls', [])
-                    self.console.print(f"[dim]Loaded conversation history: {len(self.conversation_history)} messages, {len(self.tool_calls_history)} total tool calls[/dim]")
+                    self.conversation_history = data.get("conversation_history", [])
+                    self.tool_calls_history = data.get("tool_calls_history", [])
+                    self.current_conversation_tool_calls = data.get("current_conversation_tool_calls", [])
+                    self.console.print(
+                        f"[dim]Loaded conversation history: {len(self.conversation_history)} messages, {len(self.tool_calls_history)} total tool calls[/dim]"
+                    )
         except Exception as e:
             self.console.print(f"[yellow]Warning: Could not load conversation history: {e}[/yellow]")
             self.conversation_history = []
@@ -96,14 +98,14 @@ class Agent:
         try:
             self._ensure_sessions_dir()
             data = {
-                'session_id': self.session_id,
-                'agent_name': self.name,
-                'created_at': datetime.now().isoformat(),
-                'conversation_history': self.conversation_history,
-                'tool_calls_history': self.tool_calls_history,
-                'current_conversation_tool_calls': self.current_conversation_tool_calls,
+                "session_id": self.session_id,
+                "agent_name": self.name,
+                "created_at": datetime.now().isoformat(),
+                "conversation_history": self.conversation_history,
+                "tool_calls_history": self.tool_calls_history,
+                "current_conversation_tool_calls": self.current_conversation_tool_calls,
             }
-            with open(self.session_file, 'w', encoding='utf-8') as f:
+            with open(self.session_file, "w", encoding="utf-8") as f:
                 json.dump(data, f, indent=2, ensure_ascii=False)
         except Exception as e:
             self.console.print(f"[yellow]Warning: Could not save conversation history: {e}[/yellow]")
@@ -119,15 +121,15 @@ class Agent:
         except Exception as e:
             self.console.print(f"[yellow]Warning: Could not delete session file: {e}[/yellow]")
 
-    def get_conversation_summary(self) -> Dict[str, Any]:
+    def get_conversation_summary(self) -> dict[str, Any]:
         """Get summary of the conversation."""
         return {
-            'session_id': self.session_id,
-            'agent_name': self.name,
-            'message_count': len(self.conversation_history),
-            'tool_calls_count': len(self.tool_calls_history),
-            'current_conversation_tool_calls': len(self.current_conversation_tool_calls),
-            'session_file': str(self.session_file),
+            "session_id": self.session_id,
+            "agent_name": self.name,
+            "message_count": len(self.conversation_history),
+            "tool_calls_count": len(self.tool_calls_history),
+            "current_conversation_tool_calls": len(self.current_conversation_tool_calls),
+            "session_file": str(self.session_file),
         }
 
     @property
@@ -138,7 +140,7 @@ class Agent:
     def max_steps(self) -> int:
         return self.agent_spec.max_steps
 
-    def _get_system_prompt(self, work_dir: Optional[Path] = None) -> str:
+    def _get_system_prompt(self, work_dir: Path | None = None) -> str:
         """Get resolved system prompt with template variables."""
         current_time = datetime.now().isoformat()
         work_dir_str = str(work_dir.resolve()) if work_dir else str(Path.cwd())
@@ -159,7 +161,7 @@ class Agent:
             self.console.print(f"[yellow]Warning: Missing template variable {e}[/yellow]")
             return self.agent_spec.system_prompt
 
-    async def _get_mcp_tools_info(self, cfg) -> List[Dict[str, Any]]:
+    async def _get_mcp_tools_info(self, cfg) -> list[dict[str, Any]]:
         """Get information about available MCP tools."""
         tools_info = []
 
@@ -167,11 +169,13 @@ class Agent:
             try:
                 tools = await list_mcp_tools(server)
                 for tool in tools:
-                    tools_info.append({
-                        "name": f"mcp.{server_name}.{tool['name']}",
-                        "description": tool.get("description", ""),
-                        "server": server_name,
-                    })
+                    tools_info.append(
+                        {
+                            "name": f"mcp.{server_name}.{tool['name']}",
+                            "description": tool.get("description", ""),
+                            "server": server_name,
+                        }
+                    )
             except Exception as e:
                 self.console.print(f"[yellow]Warning: Could not load tools from {server_name}: {e}[/yellow]")
 
@@ -180,7 +184,9 @@ class Agent:
     def _should_limit_tool_calls(self, tool_name: str) -> bool:
         """Check if a tool should be limited to prevent infinite loops."""
         # Per-tool limits (each tool tracked separately)
-        current_conversation_calls = sum(1 for call in self.current_conversation_tool_calls if call.get("tool") == tool_name)
+        current_conversation_calls = sum(
+            1 for call in self.current_conversation_tool_calls if call.get("tool") == tool_name
+        )
 
         # read_file: 10 per conversation, 600 per session
         if tool_name == "read_file":
@@ -209,7 +215,7 @@ class Agent:
         """Add context information for tool execution."""
         self.execution_context[key] = value
 
-    def _get_context_vars(self) -> Dict[str, str]:
+    def _get_context_vars(self) -> dict[str, str]:
         """Get context variables for system prompt."""
         return {
             "step_count": str(self.step_count),
@@ -219,11 +225,7 @@ class Agent:
         }
 
     async def run(
-        self,
-        user_input: str,
-        work_dir: Optional[Path] = None,
-        stream: bool = True,
-        show_progress: bool = True
+        self, user_input: str, work_dir: Path | None = None, stream: bool = True, show_progress: bool = True
     ) -> str:
         """
         Run the agent with the given user input.
@@ -254,7 +256,11 @@ class Agent:
                 messages = [{"role": "system", "content": system_prompt}]
 
                 # Add conversation history (limit to last 20 messages to avoid context overflow)
-                history_to_add = self.conversation_history[-20:] if len(self.conversation_history) > 20 else self.conversation_history
+                history_to_add = (
+                    self.conversation_history[-20:]
+                    if len(self.conversation_history) > 20
+                    else self.conversation_history
+                )
                 messages.extend(history_to_add)
 
                 # Add current user input
@@ -266,11 +272,7 @@ class Agent:
 
                 # Run chat with tools
                 result = await self._run_with_tools(
-                    messages=messages,
-                    tools=tools,
-                    tool_registry=tool_registry,
-                    stream=stream,
-                    status=status
+                    messages=messages, tools=tools, tool_registry=tool_registry, stream=stream, status=status
                 )
 
                 # Save conversation exchange
@@ -286,16 +288,17 @@ class Agent:
             self.console.print(f"[red]Agent execution failed:[/red] {e}")
             raise AgentExecutionError(f"Agent execution failed: {e}") from e
 
-    async def _build_tools(self) -> List[Dict[str, Any]]:
+    async def _build_tools(self) -> list[dict[str, Any]]:
         """Build list of available tools."""
         tools = []
 
         # Add all built-in tools from registry
         from .tools import get_tool_registry
+
         registry = get_tool_registry()
         for tool_name in registry.list_tools():
             tool = registry.get_tool(tool_name)
-            if tool and hasattr(tool, 'get_spec'):
+            if tool and hasattr(tool, "get_spec"):
                 tools.append(tool.get_spec())
 
         # Load MCP tools
@@ -321,20 +324,22 @@ class Agent:
                     tname = info.get("name") or "tool"
                     oname = f"mcp.{name}.{tname}"
                     params = info.get("inputSchema") or {"type": "object"}
-                    tools.append({
-                        "type": "function",
-                        "function": {
-                            "name": oname,
-                            "description": info.get("description", ""),
-                            "parameters": params
-                        },
-                    })
+                    tools.append(
+                        {
+                            "type": "function",
+                            "function": {
+                                "name": oname,
+                                "description": info.get("description", ""),
+                                "parameters": params,
+                            },
+                        }
+                    )
             except Exception as e:
                 self.console.print(f"[yellow]MCP tool discovery failed for server {name}:[/yellow] {e}")
 
         return tools
 
-    async def _build_tool_registry(self) -> Dict[str, Any]:
+    async def _build_tool_registry(self) -> dict[str, Any]:
         """Build tool registry for MCP tool dispatch."""
         registry = {}
 
@@ -360,12 +365,12 @@ class Agent:
                     tname = info.get("name") or "tool"
                     oname = f"mcp.{name}.{tname}"
                     registry[oname] = (name, tname)
-            except Exception as e:
+            except Exception:
                 pass  # Already logged in _build_tools
 
         return registry
 
-    def _get_read_file_tool_spec(self) -> Dict[str, Any]:
+    def _get_read_file_tool_spec(self) -> dict[str, Any]:
         """Get read_file tool specification."""
         return {
             "type": "function",
@@ -377,18 +382,18 @@ class Agent:
                     "properties": {
                         "path": {
                             "type": "string",
-                            "description": "Path to a specific FILE (not directory). Use relative paths like 'src/amcp/readfile.py', NEVER directories like 'src/amcp'. COMMON FILES: 'src/amcp/readfile.py', 'src/amcp/rg.py', 'src/amcp/cli.py', 'src/amcp/chat.py', 'README.md', 'pyproject.toml'. Always include the file extension (.py, .md, .toml, etc)."
+                            "description": "Path to a specific FILE (not directory). Use relative paths like 'src/amcp/readfile.py', NEVER directories like 'src/amcp'. COMMON FILES: 'src/amcp/readfile.py', 'src/amcp/rg.py', 'src/amcp/cli.py', 'src/amcp/chat.py', 'README.md', 'pyproject.toml'. Always include the file extension (.py, .md, .toml, etc).",
                         },
                         "ranges": {
                             "type": "array",
                             "items": {"type": "string", "pattern": "^\\d+-\\d+$"},
-                            "description": "Optional list of line ranges like '1-200'. Use only if you need specific line ranges. For general file analysis, omit this to get the full file."
+                            "description": "Optional list of line ranges like '1-200'. Use only if you need specific line ranges. For general file analysis, omit this to get the full file.",
                         },
                         "max_lines": {
                             "type": "integer",
                             "minimum": 1,
                             "maximum": 5000,
-                            "description": "Safety cap for lines returned per block (default 400)"
+                            "description": "Safety cap for lines returned per block (default 400)",
                         },
                     },
                     "required": ["path"],
@@ -399,11 +404,11 @@ class Agent:
 
     async def _run_with_tools(
         self,
-        messages: List[Dict[str, Any]],
-        tools: List[Dict[str, Any]],
-        tool_registry: Dict[str, Any],
+        messages: list[dict[str, Any]],
+        tools: list[dict[str, Any]],
+        tool_registry: dict[str, Any],
         stream: bool,
-        status: Status
+        status: Status,
     ) -> str:
         """Run chat with tools and enhanced tracking."""
         cfg = load_config()
@@ -420,22 +425,22 @@ class Agent:
             tools=tools,
             tool_registry=tool_registry,
             stream=stream,
-            status=status
+            status=status,
         )
 
     async def _enhanced_chat_with_tools(
         self,
         client,
         model: str,
-        messages: List[Dict[str, Any]],
-        tools: List[Dict[str, Any]],
-        tool_registry: Dict[str, Any],
+        messages: list[dict[str, Any]],
+        tools: list[dict[str, Any]],
+        tool_registry: dict[str, Any],
         stream: bool,
         status: Status,
-        max_steps: Optional[int] = None
+        max_steps: int | None = None,
     ) -> str:
         """Enhanced version of _chat_with_tools with better tracking."""
-        from .chat import _get_chat_runtime_settings, _dispatch_tool_call
+        from .chat import _get_chat_runtime_settings
 
         max_steps = max_steps or self.max_steps
 
@@ -471,13 +476,17 @@ class Agent:
                         limited_tools.append(tool_name)
 
                 if limited_tools:
-                    status.update(f"[bold]Agent {self.name}[/bold] - Tools {limited_tools} limited, forcing response...")
+                    status.update(
+                        f"[bold]Agent {self.name}[/bold] - Tools {limited_tools} limited, forcing response..."
+                    )
                     self.console.print(f"[yellow]Tools {limited_tools} limited, forcing response[/yellow]")
                     # Add system message to force response
-                    messages.append({
-                        "role": "system",
-                        "content": f"You have already called the following tools too many times: {', '.join(limited_tools)}. Please analyze the information you have and provide your response without calling these tools again."
-                    })
+                    messages.append(
+                        {
+                            "role": "system",
+                            "content": f"You have already called the following tools too many times: {', '.join(limited_tools)}. Please analyze the information you have and provide your response without calling these tools again.",
+                        }
+                    )
                     # Get a final response from the LLM with the current messages
                     try:
                         final_resp = client.chat.completions.create(
@@ -529,11 +538,13 @@ class Agent:
                                             parts.append(c.get("text", ""))
                                     tool_result_text = "\\n\\n".join(parts) or json.dumps(mcp_resp, ensure_ascii=False)
 
-                                    self.console.print(Panel(
-                                        f"✅ MCP tool {tool_name} executed successfully",
-                                        title="Tool Result",
-                                        border_style="green"
-                                    ))
+                                    self.console.print(
+                                        Panel(
+                                            f"✅ MCP tool {tool_name} executed successfully",
+                                            title="Tool Result",
+                                            border_style="green",
+                                        )
+                                    )
                                 else:
                                     tool_result_text = f"Error: Unknown MCP server {server_name}"
                             else:
@@ -541,6 +552,7 @@ class Agent:
                         else:
                             # Handle built-in tools
                             from .tools import get_tool_registry
+
                             registry = get_tool_registry()
                             args = json.loads(tc.function.arguments or "{}")
 
@@ -549,42 +561,48 @@ class Agent:
                             if tool_result.success:
                                 tool_result_text = tool_result.content
                                 preview = tool_result_text[:200] if len(tool_result_text) > 200 else tool_result_text
-                                self.console.print(Panel(
-                                    preview,
-                                    title=f"Tool: {tool_name}",
-                                    border_style="blue"
-                                ))
+                                self.console.print(Panel(preview, title=f"Tool: {tool_name}", border_style="blue"))
                             else:
                                 tool_result_text = f"Error: {tool_result.error}"
-                                self.console.print(Panel(
-                                    tool_result_text,
-                                    title=f"Tool Error: {tool_name}",
-                                    border_style="red"
-                                ))
+                                self.console.print(
+                                    Panel(tool_result_text, title=f"Tool Error: {tool_name}", border_style="red")
+                                )
 
                         # Add tool result to messages
-                        messages.append({
-                            "role": "assistant",
-                            "content": msg.content or "",
-                            "tool_calls": [{"id": tc.id, "type": "function", "function": {"name": tool_name, "arguments": tc.function.arguments or "{}"}}]
-                        })
-                        messages.append({
-                            "role": "tool",
-                            "tool_call_id": tc.id,
-                            "name": tool_name,
-                            "content": tool_result_text,
-                        })
+                        messages.append(
+                            {
+                                "role": "assistant",
+                                "content": msg.content or "",
+                                "tool_calls": [
+                                    {
+                                        "id": tc.id,
+                                        "type": "function",
+                                        "function": {"name": tool_name, "arguments": tc.function.arguments or "{}"},
+                                    }
+                                ],
+                            }
+                        )
+                        messages.append(
+                            {
+                                "role": "tool",
+                                "tool_call_id": tc.id,
+                                "name": tool_name,
+                                "content": tool_result_text,
+                            }
+                        )
 
                     except Exception as e:
                         error_msg = f"Tool {tool_name} error: {type(e).__name__}: {e}"
                         self.console.print(f"[red]{error_msg}[/red]")
 
-                        messages.append({
-                            "role": "tool",
-                            "tool_call_id": tc.id,
-                            "name": tool_name,
-                            "content": error_msg,
-                        })
+                        messages.append(
+                            {
+                                "role": "tool",
+                                "tool_call_id": tc.id,
+                                "name": tool_name,
+                                "content": error_msg,
+                            }
+                        )
 
                 continue
             else:
@@ -608,7 +626,7 @@ class Agent:
         else:
             return typer.ctx.obj or typer.Context.NULL
 
-    def get_execution_summary(self) -> Dict[str, Any]:
+    def get_execution_summary(self) -> dict[str, Any]:
         """Get summary of agent execution."""
         return {
             "agent_name": self.name,

src/amcp/agent_spec.py

@@ -1,10 +1,9 @@
 from __future__ import annotations
 
-import yaml
-from dataclasses import dataclass, field
+from dataclasses import dataclass
 from pathlib import Path
-from typing import Any, Dict, List
 
+import yaml
 from pydantic import BaseModel, Field
 
 from .config import load_config as load_app_config
@@ -12,29 +11,24 @@ from .config import load_config as load_app_config
 
 class AgentSpecError(Exception):
     """Raised when agent specification is invalid."""
+
     pass
 
 
 class AgentSpec(BaseModel):
     """Agent specification model."""
-    
+
     name: str = Field(description="Agent name")
     description: str = Field(default="", description="Agent description")
     system_prompt: str = Field(description="System prompt for the agent")
-    system_prompt_template: str = Field(
-        default="", 
-        description="System prompt template with variables"
-    )
-    system_prompt_vars: Dict[str, str] = Field(
-        default_factory=dict, 
-        description="Variables for system prompt template"
-    )
-    tools: List[str] = Field(default_factory=list, description="Available tools")
-    exclude_tools: List[str] = Field(default_factory=list, description="Tools to exclude")
+    system_prompt_template: str = Field(default="", description="System prompt template with variables")
+    system_prompt_vars: dict[str, str] = Field(default_factory=dict, description="Variables for system prompt template")
+    tools: list[str] = Field(default_factory=list, description="Available tools")
+    exclude_tools: list[str] = Field(default_factory=list, description="Tools to exclude")
     max_steps: int = Field(default=5, description="Maximum tool execution steps")
     model: str = Field(default="", description="Preferred model name")
     base_url: str = Field(default="", description="Preferred base URL")
-    
+
     class Config:
         extra = "allow"
 
@@ -42,12 +36,12 @@ class AgentSpec(BaseModel):
 @dataclass
 class ResolvedAgentSpec:
     """Resolved agent specification with all defaults applied."""
-    
+
     name: str
     description: str
     system_prompt: str
-    tools: List[str]
-    exclude_tools: List[str]
+    tools: list[str]
+    exclude_tools: list[str]
     max_steps: int
     model: str
     base_url: str
@@ -56,45 +50,45 @@ class ResolvedAgentSpec:
 def load_agent_spec(agent_file: Path) -> ResolvedAgentSpec:
     """
     Load agent specification from YAML file.
-    
+
     Args:
         agent_file: Path to agent specification file
-        
+
     Returns:
         Resolved agent specification
-        
+
     Raises:
         AgentSpecError: If file is invalid or cannot be loaded
     """
     if not agent_file.exists():
         raise AgentSpecError(f"Agent spec file not found: {agent_file}")
-    
+
     try:
-        with open(agent_file, 'r', encoding='utf-8') as f:
+        with open(agent_file, encoding="utf-8") as f:
             data = yaml.safe_load(f)
     except yaml.YAMLError as e:
         raise AgentSpecError(f"Invalid YAML in agent spec file: {e}") from e
-    
+
     if not data:
         raise AgentSpecError(f"Empty agent spec file: {agent_file}")
-    
+
     try:
         spec = AgentSpec(**data)
     except Exception as e:
         raise AgentSpecError(f"Invalid agent spec format: {e}") from e
-    
+
     # Apply defaults from global config
     cfg = load_app_config()
     default_model = cfg.chat.model if cfg.chat and cfg.chat.model else ""
     default_base_url = cfg.chat.base_url if cfg.chat and cfg.chat.base_url else ""
-    
+
     # Resolve system prompt
     system_prompt = spec.system_prompt
     if spec.system_prompt_template and spec.system_prompt_vars:
         system_prompt = spec.system_prompt_template.format(**spec.system_prompt_vars)
     elif spec.system_prompt_template:
         system_prompt = spec.system_prompt_template
-    
+
     return ResolvedAgentSpec(
         name=spec.name,
         description=spec.description,
@@ -134,18 +128,18 @@ Current time: {current_time}""",
         exclude_tools=[],
         max_steps=10,
         model="",
-        base_url=""
+        base_url="",
     )
 
 
-def list_available_agents(agents_dir: Path) -> List[Path]:
+def list_available_agents(agents_dir: Path) -> list[Path]:
     """List all available agent specification files."""
     if not agents_dir.exists():
         return []
-    
+
     agent_files = []
     for file_path in agents_dir.rglob("*.yaml"):
         if file_path.is_file():
             agent_files.append(file_path)
-    
+
     return sorted(agent_files)

src/amcp/chat.py

@@ -1,19 +1,18 @@
 from __future__ import annotations
 
-import os
-import sys
 import json
-from typing import Annotated, Optional, Iterable
+import os
+import re
+from collections.abc import Iterable
+from pathlib import Path
 
 from rich.console import Console
-from rich.panel import Panel
 from rich.live import Live
 from rich.markdown import Markdown
+from rich.panel import Panel
 
-from .mcp_client import call_mcp_tool, list_mcp_tools
 from .config import AMCPConfig, ChatConfig, load_config
-from pathlib import Path
-import re
+from .mcp_client import call_mcp_tool, list_mcp_tools
 from .readfile import read_file_with_ranges
 
 console = Console()
@@ -24,9 +23,9 @@ def _run_quietly(coro):
     This helps hide noisy MCP server startup logs in chat mode.
     """
     import contextlib
-    with open(os.devnull, "w") as devnull:
-        with contextlib.redirect_stderr(devnull):
-            return __import__("asyncio").run(coro)
+
+    with open(os.devnull, "w") as devnull, contextlib.redirect_stderr(devnull):
+        return __import__("asyncio").run(coro)
 
 
 def _resolve_base_url(cli_base: str | None, cfg: ChatConfig | None) -> str:
@@ -42,7 +41,7 @@ def _resolve_base_url(cli_base: str | None, cfg: ChatConfig | None) -> str:
     return base
 
 
-def _resolve_api_key(cli_key: Optional[str], cfg: ChatConfig | None) -> Optional[str]:
+def _resolve_api_key(cli_key: str | None, cfg: ChatConfig | None) -> str | None:
     # CLI > config > env
     if cli_key:
         return cli_key
@@ -51,10 +50,10 @@ def _resolve_api_key(cli_key: Optional[str], cfg: ChatConfig | None) -> Optional
     return os.environ.get("SAMBANOVA_API_KEY") or os.environ.get("OPENAI_API_KEY")
 
 
-def _make_client(base_url: str, api_key: Optional[str]):
+def _make_client(base_url: str, api_key: str | None):
     try:
         from openai import OpenAI
-    except Exception as e:  # pragma: no cover
+    except Exception:  # pragma: no cover
         console.print("[red]openai package not installed. Please install dependencies.[/red]")
         raise
     return OpenAI(base_url=base_url, api_key=api_key or "")
@@ -102,7 +101,7 @@ def _stream_chat(client, model: str, messages: list[dict], stream: bool = True)
                     if item.type == "output_text":
                         parts.append(item.text)
                 return "".join(parts)
-        except Exception as e:
+        except Exception:
             raise
 
 
@@ -132,13 +131,18 @@ def _attach_file_context(path: Path, ranges: Iterable[str] | None, max_lines: in
         lines = b["lines"]
         if not ranges and len(lines) > max_lines:
             lines = lines[:max_lines]
-        llm_context.append(f"FILE: {path} ({b['start']}-{b['start'] + len(lines) - 1})\n" + "\n".join(l for _, l in lines))
+        llm_context.append(
+            f"FILE: {path} ({b['start']}-{b['start'] + len(lines) - 1})\n" + "\n".join(l for _, l in lines)
+        )
         if not ranges and len(b["lines"]) > max_lines:
             llm_context.append("[...truncated...]")
     return "\n\n".join(rendered_parts), "\n\n".join(llm_context)
 
 
-_READ_CMD = re.compile(r"^\s*(?:/read|read|open|查看|读取|打开)\s+(?P<path>\S+)(?:\s+(?:lines?|行|第)\s*(?P<s>\d+)\s*[-~]\s*(?P<e>\d+))?\s*$", re.I)
+_READ_CMD = re.compile(
+    r"^\s*(?:/read|read|open|查看|读取|打开)\s+(?P<path>\S+)(?:\s+(?:lines?|行|第)\s*(?P<s>\d+)\s*[-~]\s*(?P<e>\d+))?\s*$",
+    re.I,
+)
 _PATH_RANGE_INLINE = re.compile(r"(?P<path>\S+?):(?P<s>\d+)-(?P<e>\d+)")
 
 
@@ -175,13 +179,21 @@ def _builtin_read_tool_spec() -> dict:
             "parameters": {
                 "type": "object",
                 "properties": {
-                        "path": {"type": "string", "description": "Path to a specific FILE (not directory). Use relative paths like 'src/amcp/readfile.py', NEVER directories like 'src/amcp'. COMMON FILES: 'src/amcp/readfile.py', 'src/amcp/rg.py', 'src/amcp/cli.py', 'src/amcp/chat.py', 'README.md', 'pyproject.toml'. Always include the file extension (.py, .md, .toml, etc)."},
-                        "ranges": {
-                            "type": "array",
-                            "items": {"type": "string", "pattern": "^\\d+-\\d+$"},
-                            "description": "Optional list of line ranges like '1-200'. Use only if you need specific line ranges. For general file analysis, omit this to get the full file."
-                        },
-                        "max_lines": {"type": "integer", "minimum": 1, "maximum": 5000, "description": "Safety cap for lines returned per block (default 400)"},
+                    "path": {
+                        "type": "string",
+                        "description": "Path to a specific FILE (not directory). Use relative paths like 'src/amcp/readfile.py', NEVER directories like 'src/amcp'. COMMON FILES: 'src/amcp/readfile.py', 'src/amcp/rg.py', 'src/amcp/cli.py', 'src/amcp/chat.py', 'README.md', 'pyproject.toml'. Always include the file extension (.py, .md, .toml, etc).",
+                    },
+                    "ranges": {
+                        "type": "array",
+                        "items": {"type": "string", "pattern": "^\\d+-\\d+$"},
+                        "description": "Optional list of line ranges like '1-200'. Use only if you need specific line ranges. For general file analysis, omit this to get the full file.",
+                    },
+                    "max_lines": {
+                        "type": "integer",
+                        "minimum": 1,
+                        "maximum": 5000,
+                        "description": "Safety cap for lines returned per block (default 400)",
+                    },
                 },
                 "required": ["path"],
                 "additionalProperties": False,
@@ -193,8 +205,8 @@ def _builtin_read_tool_spec() -> dict:
 def _get_chat_runtime_settings(override: dict | None = None) -> dict:
     cfg: AMCPConfig = load_config()
     chat_cfg = cfg.chat
-    tool_loop_limit = (chat_cfg.tool_loop_limit if chat_cfg and chat_cfg.tool_loop_limit else 5)
-    default_max_lines = (chat_cfg.default_max_lines if chat_cfg and chat_cfg.default_max_lines else 400)
+    tool_loop_limit = chat_cfg.tool_loop_limit if chat_cfg and chat_cfg.tool_loop_limit else 5
+    default_max_lines = chat_cfg.default_max_lines if chat_cfg and chat_cfg.default_max_lines else 400
     roots: list[Path]
     if chat_cfg and chat_cfg.read_roots:
         roots = [Path(r).expanduser().resolve() for r in chat_cfg.read_roots]
@@ -225,12 +237,14 @@ def _dispatch_tool_call(name: str, arguments: dict, *, settings: dict) -> tuple[
     allowed_roots: list[Path] = settings["allowed_roots"]
     if not any(_is_within_root(p, root) for root in allowed_roots):
         raise ValueError(f"Path {p} is outside allowed roots: {allowed_roots}")
-    
+
     # Check if path exists and is a file
     if not p.exists():
         raise FileNotFoundError(f"File not found: {p}")
     if not p.is_file():
-        raise ValueError(f"Path is a directory, not a file: {p}. Use a specific file like 'src/amcp/readfile.py' instead of just 'src/amcp'.")
+        raise ValueError(
+            f"Path is a directory, not a file: {p}. Use a specific file like 'src/amcp/readfile.py' instead of just 'src/amcp'."
+        )
 
     rendered, llm = _attach_file_context(p, ranges, max_lines=max_lines)
     # content for model
@@ -246,7 +260,9 @@ def _is_within_root(path: Path, root: Path) -> bool:
         return False
 
 
-def _build_mcp_tools_and_registry(cfg: AMCPConfig, chat_cfg: ChatConfig | None, servers_override: list[str] | None) -> tuple[list[dict], dict]:
+def _build_mcp_tools_and_registry(
+    cfg: AMCPConfig, chat_cfg: ChatConfig | None, servers_override: list[str] | None
+) -> tuple[list[dict], dict]:
     # Decide which servers to include
     if chat_cfg and chat_cfg.mcp_tools_enabled is False:
         return [], {}
@@ -268,22 +284,33 @@ def _build_mcp_tools_and_registry(cfg: AMCPConfig, chat_cfg: ChatConfig | None,
                 oname = f"mcp.{name}.{tname}"
                 # Parameters: best effort; prefer server-provided schema when available
                 params = info.get("inputSchema") or {"type": "object"}
-                tools.append({
-                    "type": "function",
-                    "function": {"name": oname, "description": info.get("description", ""), "parameters": params},
-                })
+                tools.append(
+                    {
+                        "type": "function",
+                        "function": {"name": oname, "description": info.get("description", ""), "parameters": params},
+                    }
+                )
                 reg[oname] = (name, tname)
         except Exception as e:
             console.print(f"[yellow]MCP tool discovery failed for server {name}:[/yellow] {e}")
     return tools, reg
 
 
-def _chat_with_tools(client, model: str, base_messages: list[dict], stream: bool, settings_override: dict | None = None, *, extra_tools: list[dict] | None = None, tool_registry: dict | None = None) -> str:
+def _chat_with_tools(
+    client,
+    model: str,
+    base_messages: list[dict],
+    stream: bool,
+    settings_override: dict | None = None,
+    *,
+    extra_tools: list[dict] | None = None,
+    tool_registry: dict | None = None,
+) -> str:
     messages = list(base_messages)
     used_tools = False
     settings = _get_chat_runtime_settings(settings_override)
     max_steps = settings["tool_loop_limit"]
-    tools = [ _builtin_read_tool_spec() ]
+    tools = [_builtin_read_tool_spec()]
     if extra_tools:
         tools.extend(extra_tools)
     registry = tool_registry or {}
@@ -309,10 +336,12 @@ def _chat_with_tools(client, model: str, base_messages: list[dict], stream: bool
                     read_file_call_count += 1
                     if read_file_call_count >= 2:
                         # Force the model to respond by adding a system message
-                        messages.append({
-                            "role": "system", 
-                            "content": "You have already read the file content. Please analyze the information you have and provide your response without calling the read_file tool again."
-                        })
+                        messages.append(
+                            {
+                                "role": "system",
+                                "content": "You have already read the file content. Please analyze the information you have and provide your response without calling the read_file tool again.",
+                            }
+                        )
                         break
             # append assistant message with tool calls
             assistant_msg = {"role": "assistant", "content": msg.content or "", "tool_calls": []}
@@ -323,11 +352,13 @@ def _chat_with_tools(client, model: str, base_messages: list[dict], stream: bool
                     args = json.loads(fn.arguments or "{}")
                 except Exception:
                     pass
-                assistant_msg["tool_calls"].append({
-                    "id": tc.id,
-                    "type": "function",
-                    "function": {"name": fn.name, "arguments": fn.arguments or "{}"},
-                })
+                assistant_msg["tool_calls"].append(
+                    {
+                        "id": tc.id,
+                        "type": "function",
+                        "function": {"name": fn.name, "arguments": fn.arguments or "{}"},
+                    }
+                )
                 try:
                     if fn.name.startswith("mcp.") and registry:
                         server_name, inner_name = registry.get(fn.name, (None, None))
@@ -360,15 +391,19 @@ def _chat_with_tools(client, model: str, base_messages: list[dict], stream: bool
                     console.print(f"[red]Tool {fn.name} error:[/red] {e}")
                     # Add more specific error info for TaskGroup issues
                     if "TaskGroup" in str(e) and "exa" in fn.name:
-                        console.print("[yellow]Hint: Exa MCP server may be experiencing connectivity issues. Try rephrasing your request or using local tools instead.[/yellow]")
+                        console.print(
+                            "[yellow]Hint: Exa MCP server may be experiencing connectivity issues. Try rephrasing your request or using local tools instead.[/yellow]"
+                        )
                 # append tool result
                 messages.append(assistant_msg)
-                messages.append({
-                    "role": "tool",
-                    "tool_call_id": tc.id,
-                    "name": fn.name,
-                    "content": tool_result_text,
-                })
+                messages.append(
+                    {
+                        "role": "tool",
+                        "tool_call_id": tc.id,
+                        "name": fn.name,
+                        "content": tool_result_text,
+                    }
+                )
             continue  # back to the loop
         else:
             final_text = msg.content or ""
@@ -380,6 +415,7 @@ def _chat_with_tools(client, model: str, base_messages: list[dict], stream: bool
             return final_text
     return "[Tool loop limit reached]"
 
+
 def _normalize_exa_web_search_args(args: dict) -> dict:
     out = dict(args or {})
     # Map possible synonyms
@@ -406,11 +442,17 @@ def do_exa_search(server_name: str, query: str, num_results: int = 4) -> str:
         raise RuntimeError(f"Unknown MCP server '{server_name}'. Use 'amcp mcp tools -s ...' to verify.")
     result = console.status("Calling MCP web_search_exa...")
     with result:
-        resp = __import__("asyncio").run(call_mcp_tool(cfg.servers[server_name], "web_search_exa", {
-            "query": query,
-            "numResults": num_results,
-            "type": "fast",
-        }))
+        resp = __import__("asyncio").run(
+            call_mcp_tool(
+                cfg.servers[server_name],
+                "web_search_exa",
+                {
+                    "query": query,
+                    "numResults": num_results,
+                    "type": "fast",
+                },
+            )
+        )
     # Render
     out_lines = [f"MCP search results for: {query}"]
     for block in resp.get("content", []):
@@ -424,16 +466,16 @@ cfg_global: AMCPConfig = load_config()
 
 
 def chat_once(
-    model: Optional[str],
+    model: str | None,
     user_text: str,
-    base_url: Optional[str] = None,
-    system_prompt: Optional[str] = None,
+    base_url: str | None = None,
+    system_prompt: str | None = None,
     mcp_server: str = "exa",
     stream: bool = True,
-    api_key: Optional[str] = None,
-    work_dir: Optional[Path] = None,
-    mcp_servers_override: Optional[list[str]] = None,
-    mcp_tools_enabled: Optional[bool] = None,
+    api_key: str | None = None,
+    work_dir: Path | None = None,
+    mcp_servers_override: list[str] | None = None,
+    mcp_tools_enabled: bool | None = None,
 ) -> str:
     cfg: AMCPConfig = load_config()
     chat_cfg = cfg.chat
@@ -451,36 +493,44 @@ def chat_once(
         messages.append({"role": "system", "content": system_prompt})
     messages.append({"role": "user", "content": user_text})
     overrides = {"read_roots": [str(work_dir.resolve())]} if work_dir else None
-    
+
     # Build MCP tools
     extra_tools = []
     registry = {}
     enabled = True
-    
+
     # Check if MCP tools are disabled in config
-    if hasattr(chat_cfg, 'mcp_tools_enabled') and chat_cfg and chat_cfg.mcp_tools_enabled is False:
+    if hasattr(chat_cfg, "mcp_tools_enabled") and chat_cfg and chat_cfg.mcp_tools_enabled is False:
         enabled = False
-    
+
     # Override with command line flag
     if mcp_tools_enabled is not None:
         enabled = bool(mcp_tools_enabled)
-    
+
     if enabled:
         extra_tools, registry = _build_mcp_tools_and_registry(cfg, chat_cfg, mcp_servers_override)
-    
-    return _chat_with_tools(client, resolved_model, messages, stream=stream, settings_override=overrides, extra_tools=extra_tools, tool_registry=registry)
+
+    return _chat_with_tools(
+        client,
+        resolved_model,
+        messages,
+        stream=stream,
+        settings_override=overrides,
+        extra_tools=extra_tools,
+        tool_registry=registry,
+    )
 
 
 def chat_repl(
-    model: Optional[str],
-    base_url: Optional[str] = None,
-    system_prompt: Optional[str] = None,
+    model: str | None,
+    base_url: str | None = None,
+    system_prompt: str | None = None,
     mcp_server: str = "exa",
     stream: bool = True,
-    api_key: Optional[str] = None,
-    work_dir: Optional[Path] = None,
-    mcp_servers_override: Optional[list[str]] = None,
-    mcp_tools_enabled: Optional[bool] = None,
+    api_key: str | None = None,
+    work_dir: Path | None = None,
+    mcp_servers_override: list[str] | None = None,
+    mcp_tools_enabled: bool | None = None,
 ) -> None:
     cfg: AMCPConfig = load_config()
     chat_cfg = cfg.chat
@@ -501,7 +551,7 @@ def chat_repl(
     cfg = load_config()
     chat_cfg = cfg.chat
     enabled = True
-    if hasattr(chat_cfg, 'mcp_tools_enabled') and chat_cfg and chat_cfg.mcp_tools_enabled is False:
+    if hasattr(chat_cfg, "mcp_tools_enabled") and chat_cfg and chat_cfg.mcp_tools_enabled is False:
         enabled = False
     if mcp_tools_enabled is not None:
         enabled = bool(mcp_tools_enabled)
@@ -509,19 +559,21 @@ def chat_repl(
     registry = {}
     if enabled:
         extra_tools, registry = _build_mcp_tools_and_registry(cfg, chat_cfg, mcp_servers_override)
-    enabled_servers = sorted(set(name.split('.')[1] for name in registry.keys())) if registry else []
+    enabled_servers = sorted(set(name.split(".")[1] for name in registry.keys())) if registry else []
     mcp_line = f"MCP tools: {'on' if extra_tools else 'off'}; servers: {', '.join(enabled_servers) if enabled_servers else '-'}"
-    console.print(Panel(
-        f"Chat model: [bold]{resolved_model}[/bold]\n"
-        f"Base: {base}\n"
-        f"Tool loop limit: {settings['tool_loop_limit']}\n"
-        f"Default max lines: {settings['default_max_lines']}\n"
-        f"Allowed read roots:\n{roots_str}\n"
-        f"{mcp_line}\n\n"
-        f"Commands: /read <path> [lines A-B], /search <q>, /quit",
-        title="amcp chat",
-        border_style="green",
-    ))
+    console.print(
+        Panel(
+            f"Chat model: [bold]{resolved_model}[/bold]\n"
+            f"Base: {base}\n"
+            f"Tool loop limit: {settings['tool_loop_limit']}\n"
+            f"Default max lines: {settings['default_max_lines']}\n"
+            f"Allowed read roots:\n{roots_str}\n"
+            f"{mcp_line}\n\n"
+            f"Commands: /read <path> [lines A-B], /search <q>, /quit",
+            title="amcp chat",
+            border_style="green",
+        )
+    )
 
     messages: list[dict[str, str]] = []
     if system_prompt:
@@ -562,7 +614,7 @@ def chat_repl(
                 console.print(f"[yellow]File intent parse/read warning:[/yellow] {e}")
 
         if text.startswith("/search "):
-            q = text[len("/search "):].strip()
+            q = text[len("/search ") :].strip()
             try:
                 result = do_exa_search(mcp_server, q)
                 console.print(Panel(Markdown(result), title="exa search", border_style="magenta"))
@@ -573,7 +625,15 @@ def chat_repl(
         messages.append({"role": "user", "content": text})
         try:
             extra_tools, registry = _build_mcp_tools_and_registry(load_config(), load_config().chat, None)
-            reply = _chat_with_tools(client, resolved_model, messages, stream=stream, settings_override=overrides, extra_tools=extra_tools, tool_registry=registry)
+            reply = _chat_with_tools(
+                client,
+                resolved_model,
+                messages,
+                stream=stream,
+                settings_override=overrides,
+                extra_tools=extra_tools,
+                tool_registry=registry,
+            )
             messages.append({"role": "assistant", "content": reply})
         except Exception as e:
             console.print(f"[red]Chat error:[/red] {e}")

src/amcp/cli.py

@@ -9,14 +9,13 @@ from typing import Annotated
 import typer
 from rich.console import Console
 from rich.json import JSON
-from rich.panel import Panel
 from rich.markdown import Markdown
-
-from .config import AMCPConfig, load_config, save_default_config, save_config, Server
-from .mcp_client import call_mcp_tool, list_mcp_tools
+from rich.panel import Panel
 
 from .agent import Agent
-from .agent_spec import load_agent_spec, get_default_agent_spec, list_available_agents
+from .agent_spec import get_default_agent_spec, list_available_agents, load_agent_spec
+from .config import AMCPConfig, load_config, save_default_config
+from .mcp_client import call_mcp_tool, list_mcp_tools
 
 app = typer.Typer(add_completion=False, context_settings={"help_option_names": ["-h", "--help"]})
 console = Console()
@@ -45,7 +44,7 @@ def mcp_tools(server: Annotated[str, typer.Option("--server", "-s")]):
 def mcp_call(
     server: Annotated[str, typer.Option("--server", "-s")],
     tool: Annotated[str, typer.Option("--tool", "-t")],
-    args: Annotated[str | None, typer.Option("--args", help="JSON-encoded arguments")]=None,
+    args: Annotated[str | None, typer.Option("--args", help="JSON-encoded arguments")] = None,
 ):
     cfg: AMCPConfig = load_config()
     if server not in cfg.servers:
@@ -60,35 +59,44 @@ def main(
     ctx: typer.Context,
     message: Annotated[str | None, typer.Option("--once", help="Send one message and exit")] = None,
     agent_file: Annotated[str | None, typer.Option("--agent", help="Path to agent specification file")] = None,
-    work_dir: Annotated[Path | None, typer.Option("--work-dir", "-w", help="Set working directory", exists=True, file_okay=False, dir_okay=True, readable=True)] = None,
+    work_dir: Annotated[
+        Path | None,
+        typer.Option(
+            "--work-dir", "-w", help="Set working directory", exists=True, file_okay=False, dir_okay=True, readable=True
+        ),
+    ] = None,
     no_progress: Annotated[bool, typer.Option("--no-progress", help="Disable progress indicators")] = False,
     list_agents: Annotated[bool, typer.Option("--list", help="List available agent specifications")] = False,
-    session_id: Annotated[str | None, typer.Option("--session", help="Use specific session ID for conversation continuity")] = None,
+    session_id: Annotated[
+        str | None, typer.Option("--session", help="Use specific session ID for conversation continuity")
+    ] = None,
     clear_session: Annotated[bool, typer.Option("--clear", help="Clear conversation history for the session")] = False,
-    list_sessions: Annotated[bool, typer.Option("--list-sessions", help="List available conversation sessions")] = False,
+    list_sessions: Annotated[
+        bool, typer.Option("--list-sessions", help="List available conversation sessions")
+    ] = False,
 ) -> None:
     """Enhanced agent chat with improved tool management and context awareness."""
-    
+
     # If a subcommand is invoked, don't run the agent
     if ctx.invoked_subcommand is not None:
         return
-    
+
     # Handle session listing
     if list_sessions:
         sessions_dir = Path.home() / ".config" / "amcp" / "sessions"
         if not sessions_dir.exists():
             console.print("[yellow]No sessions directory found[/yellow]")
             return
-        
+
         session_files = list(sessions_dir.glob("*.json"))
         if not session_files:
             console.print("[yellow]No conversation sessions found[/yellow]")
             return
-        
+
         console.print("[bold]Available Conversation Sessions:[/bold]")
         for session_file in sorted(session_files, key=lambda f: f.stat().st_mtime, reverse=True):
             try:
-                with open(session_file, 'r', encoding='utf-8') as f:
+                with open(session_file, encoding="utf-8") as f:
                     data = json.load(f)
                     console.print(f"📄 {data.get('session_id', session_file.stem)}")
                     console.print(f"   Agent: {data.get('agent_name', 'Unknown')}")
@@ -99,22 +107,24 @@ def main(
             except Exception as e:
                 console.print(f"❌ {session_file.name}: {e}")
         return
-    
+
     if list_agents:
         # Check both global config dir and local agents dir
         agents_dir = Path(os.path.expanduser("~/.config/amcp/agents"))
         local_agents_dir = Path("agents")
-        
+
         agent_files = list_available_agents(agents_dir)
         local_agent_files = list_available_agents(local_agents_dir)
-        
+
         # Combine both lists
         all_agent_files = agent_files + local_agent_files
-        
+
         if not all_agent_files:
-            console.print("[yellow]No agent specifications found. Create one in ~/.config/amcp/agents/ or local agents/[/yellow]")
+            console.print(
+                "[yellow]No agent specifications found. Create one in ~/.config/amcp/agents/ or local agents/[/yellow]"
+            )
             return
-        
+
         console.print("[bold]Available Agent Specifications:[/bold]")
         for agent_file in all_agent_files:
             try:
@@ -126,7 +136,7 @@ def main(
                 console.print()
             except Exception as e:
                 console.print(f"❌ {agent_file.name}: {e}")
-        
+
         # Also show default agent
         default_spec = get_default_agent_spec()
         console.print("[bold]Default Agent:[/bold]")
@@ -134,7 +144,7 @@ def main(
         console.print(f"   Description: {default_spec.description}")
         console.print(f"   Tools: {len(default_spec.tools)}")
         return
-    
+
     try:
         # Load agent specification
         if agent_file:
@@ -147,62 +157,61 @@ def main(
         else:
             agent_spec = get_default_agent_spec()
             console.print(f"[green]Using default agent: {agent_spec.name}[/green]")
-        
+
         # Create agent with session management
         agent = Agent(agent_spec, session_id=session_id)
-        
+
         # Handle session clearing
         if clear_session:
             agent.clear_conversation_history()
             console.print(f"[green]Cleared conversation history for session: {agent.session_id}[/green]")
-        
+
         # Show session info
         session_info = agent.get_conversation_summary()
-        if session_info['message_count'] > 0:
+        if session_info["message_count"] > 0:
             console.print(f"[dim]Session {agent.session_id}: {session_info['message_count']} messages in history[/dim]")
         else:
             console.print(f"[dim]New session started: {agent.session_id}[/dim]")
-        
+
         if message is not None:
             # Single message mode
             console.print(f"[bold]🤖 Agent {agent.name}[/bold] - Processing...")
-            response = asyncio.run(agent.run(
-                user_input=message,
-                work_dir=work_dir,
-                stream=False,
-                show_progress=not no_progress
-            ))
-            
+            response = asyncio.run(
+                agent.run(user_input=message, work_dir=work_dir, stream=False, show_progress=not no_progress)
+            )
+
             console.print(Panel(Markdown(response), title=f"Agent {agent.name}", border_style="cyan"))
-            
+
             # Show execution summary
             summary = agent.get_execution_summary()
-            console.print(f"[dim]Steps: {summary['steps_taken']}/{summary['max_steps']} | Tools called: {summary['tools_called']}[/dim]")
-            
+            console.print(
+                f"[dim]Steps: {summary['steps_taken']}/{summary['max_steps']} | Tools called: {summary['tools_called']}[/dim]"
+            )
+
         else:
             # Interactive mode
             console.print(f"[bold]🤖 Agent {agent.name} - Interactive Mode[/bold]")
             console.print(f"[dim]Description: {agent_spec.description}[/dim]")
             console.print(f"[dim]Max steps: {agent_spec.max_steps} | Session: {agent.session_id}[/dim]")
-            console.print(f"[dim]Commands: 'exit' to quit, 'clear' to clear history, 'info' for session info[/dim]")
+            console.print("[dim]Commands: 'exit' to quit, 'clear' to clear history, 'info' for session info[/dim]")
             console.print()
-            
+
             while True:
                 try:
                     user_input = console.input("[bold]You:[/bold] ").strip()
-                    
-                    if user_input.lower() in ['exit', 'quit', 'q']:
+
+                    if user_input.lower() in ["exit", "quit", "q"]:
                         console.print("[green]Goodbye! 👋[/green]")
                         break
-                    
-                    if user_input.lower() == 'clear':
+
+                    if user_input.lower() == "clear":
                         agent.clear_conversation_history()
                         console.print(f"[green]Conversation history cleared for session: {agent.session_id}[/green]")
                         continue
-                    
-                    if user_input.lower() == 'info':
+
+                    if user_input.lower() == "info":
                         session_info = agent.get_conversation_summary()
-                        console.print(f"[bold]Session Info:[/bold]")
+                        console.print("[bold]Session Info:[/bold]")
                         console.print(f"Session ID: {session_info['session_id']}")
                         console.print(f"Agent: {session_info['agent_name']}")
                         console.print(f"Messages: {session_info['message_count']}")
@@ -210,25 +219,24 @@ def main(
                         console.print(f"Session file: {session_info['session_file']}")
                         console.print()
                         continue
-                    
+
                     if not user_input:
                         continue
-                    
+
                     console.print(f"[bold]🤖 Agent {agent.name}[/bold] - Processing...")
-                    response = asyncio.run(agent.run(
-                        user_input=user_input,
-                        work_dir=work_dir,
-                        stream=False,
-                        show_progress=not no_progress
-                    ))
-                    
+                    response = asyncio.run(
+                        agent.run(user_input=user_input, work_dir=work_dir, stream=False, show_progress=not no_progress)
+                    )
+
                     console.print(Panel(Markdown(response), title=f"Agent {agent.name}", border_style="cyan"))
-                    
+
                     # Show execution summary
                     summary = agent.get_execution_summary()
-                    console.print(f"[dim]Steps: {summary['steps_taken']}/{summary['max_steps']} | Tools called: {summary['tools_called']} | Session: {agent.session_id}[/dim]")
+                    console.print(
+                        f"[dim]Steps: {summary['steps_taken']}/{summary['max_steps']} | Tools called: {summary['tools_called']} | Session: {agent.session_id}[/dim]"
+                    )
                     console.print()
-                    
+
                 except EOFError:
                     console.print("[green]Goodbye! 👋[/green]")
                     break
@@ -237,31 +245,46 @@ def main(
                 except Exception as e:
                     console.print(f"[red]Error: {e}[/red]")
                     console.print()
-    
+
     except Exception as e:
         console.print(f"[red]Agent failed:[/red] {e}")
         raise typer.Exit(code=1)
 
 
 @app.command(help="Enhanced agent chat (alias for default command)")
-
 @app.command(help="Enhanced agent chat (alias for default command)")
 def agent(
     ctx: typer.Context,
     message: Annotated[str | None, typer.Option("--once", help="Send one message and exit")] = None,
     agent_file: Annotated[str | None, typer.Option("--agent", help="Path to agent specification file")] = None,
-    work_dir: Annotated[Path | None, typer.Option("--work-dir", "-w", help="Set working directory", exists=True, file_okay=False, dir_okay=True, readable=True)] = None,
+    work_dir: Annotated[
+        Path | None,
+        typer.Option(
+            "--work-dir", "-w", help="Set working directory", exists=True, file_okay=False, dir_okay=True, readable=True
+        ),
+    ] = None,
     no_progress: Annotated[bool, typer.Option("--no-progress", help="Disable progress indicators")] = False,
     list_agents: Annotated[bool, typer.Option("--list", help="List available agent specifications")] = False,
-    session_id: Annotated[str | None, typer.Option("--session", help="Use specific session ID for conversation continuity")] = None,
+    session_id: Annotated[
+        str | None, typer.Option("--session", help="Use specific session ID for conversation continuity")
+    ] = None,
     clear_session: Annotated[bool, typer.Option("--clear", help="Clear conversation history for the session")] = False,
-    list_sessions: Annotated[bool, typer.Option("--list-sessions", help="List available conversation sessions")] = False,
+    list_sessions: Annotated[
+        bool, typer.Option("--list-sessions", help="List available conversation sessions")
+    ] = False,
 ) -> None:
     """Enhanced agent chat (alias for default command)."""
-    ctx.invoke(main, message=message, agent_file=agent_file, work_dir=work_dir, 
-               no_progress=no_progress, list_agents=list_agents, session_id=session_id,
-               clear_session=clear_session, list_sessions=list_sessions)
-
+    ctx.invoke(
+        main,
+        message=message,
+        agent_file=agent_file,
+        work_dir=work_dir,
+        no_progress=no_progress,
+        list_agents=list_agents,
+        session_id=session_id,
+        clear_session=clear_session,
+        list_sessions=list_sessions,
+    )
 
 
 if __name__ == "__main__":

src/amcp/config.py

@@ -1,9 +1,9 @@
 from __future__ import annotations
 
 import os
+from collections.abc import Mapping
 from dataclasses import dataclass, field
 from pathlib import Path
-from typing import Dict, List, Mapping, Optional
 
 try:
     import tomllib  # py311+
@@ -12,7 +12,6 @@ except ModuleNotFoundError:  # pragma: no cover
 
 import tomli_w  # type: ignore
 
-
 CONFIG_DIR = Path(os.environ.get("XDG_CONFIG_HOME", Path.home() / ".config")) / "amcp"
 CONFIG_FILE = CONFIG_DIR / "config.toml"
 
@@ -20,35 +19,35 @@ CONFIG_FILE = CONFIG_DIR / "config.toml"
 @dataclass
 class Server:
     # stdio transport fields
-    command: Optional[str] = None
-    args: List[str] = field(default_factory=list)
-    env: Dict[str, str] = field(default_factory=dict)
+    command: str | None = None
+    args: list[str] = field(default_factory=list)
+    env: dict[str, str] = field(default_factory=dict)
     # http(sse) transport fields
-    url: Optional[str] = None
-    headers: Dict[str, str] = field(default_factory=dict)
+    url: str | None = None
+    headers: dict[str, str] = field(default_factory=dict)
 
 
 @dataclass
 class ChatConfig:
-    base_url: Optional[str] = None
-    model: Optional[str] = None
-    api_key: Optional[str] = None
+    base_url: str | None = None
+    model: str | None = None
+    api_key: str | None = None
     # Tool calling settings
-    tool_loop_limit: Optional[int] = None
-    default_max_lines: Optional[int] = None
-    read_roots: Optional[list[str]] = None  # list of allowed root paths for read_file
+    tool_loop_limit: int | None = None
+    default_max_lines: int | None = None
+    read_roots: list[str] | None = None  # list of allowed root paths for read_file
     # MCP tool exposure
-    mcp_tools_enabled: Optional[bool] = None
-    mcp_servers: Optional[list[str]] = None  # which servers' tools to expose; if unset, expose all configured servers
+    mcp_tools_enabled: bool | None = None
+    mcp_servers: list[str] | None = None  # which servers' tools to expose; if unset, expose all configured servers
     # Built-in file modification tools
-    write_tool_enabled: Optional[bool] = None
-    edit_tool_enabled: Optional[bool] = None
+    write_tool_enabled: bool | None = None
+    edit_tool_enabled: bool | None = None
 
 
 @dataclass
 class AMCPConfig:
-    servers: Dict[str, Server]
-    chat: Optional[ChatConfig] = None
+    servers: dict[str, Server]
+    chat: ChatConfig | None = None
 
 
 _DEFAULT = {
@@ -68,7 +67,7 @@ _DEFAULT = {
         # "mcp_servers": ["exa"]  # optional; if unset, expose all configured servers
         "write_tool_enabled": True,
         "edit_tool_enabled": True,
-    }
+    },
 }
 
 
@@ -83,7 +82,7 @@ def _decode_server(name: str, raw: Mapping[str, object]) -> Server:
     return Server(command=command_s, args=args, env=env, url=url_s, headers=headers)
 
 
-def _decode_chat(raw: Mapping[str, object] | None) -> Optional[ChatConfig]:
+def _decode_chat(raw: Mapping[str, object] | None) -> ChatConfig | None:
     if not raw:
         return None
     base_url = raw.get("base_url")

src/amcp/mcp_client.py

@@ -1,15 +1,14 @@
 from __future__ import annotations
 
 import os
-from typing import Any, Dict, List, Tuple
+from contextlib import asynccontextmanager
+from typing import Any
 
 from mcp import ClientSession
 from mcp.client.stdio import StdioServerParameters, stdio_client
-from mcp.client.sse import sse_client
 from mcp.client.streamable_http import streamablehttp_client
 
 from .config import Server
-import os
 
 
 def _expand_env(value: str) -> str:
@@ -17,13 +16,10 @@ def _expand_env(value: str) -> str:
     return os.path.expandvars(value)
 
 
-def _headers_with_env(headers: Dict[str, str]) -> Dict[str, str]:
+def _headers_with_env(headers: dict[str, str]) -> dict[str, str]:
     return {k: _expand_env(v) for k, v in headers.items()}
 
 
-from contextlib import asynccontextmanager
-
-
 @asynccontextmanager
 async def _open_transport(server: Server):
     """Yield (read, write) streams for the given server, regardless of transport."""
@@ -39,63 +35,63 @@ async def _open_transport(server: Server):
     merged = dict(base_env)
     for k, v in (server.env or {}).items():
         merged[str(k)] = _expand_env(str(v))
-    
+
     params = StdioServerParameters(command=server.command or "", args=server.args, env=merged)
-    
+
     # Suppress MCP server stderr by redirecting to devnull
-    with open(os.devnull, 'w') as devnull:
+    with open(os.devnull, "w") as devnull:
         async with stdio_client(params, errlog=devnull) as (read, write):
             yield read, write
 
 
-async def list_mcp_tools(server: Server) -> List[Dict[str, Any]]:
-    async with _open_transport(server) as (read, write):
-        async with ClientSession(read, write) as session:
-            await session.initialize()
-            result = await session.list_tools()
-            tools: List[Dict[str, Any]] = []
-            for t in result.tools:
+async def list_mcp_tools(server: Server) -> list[dict[str, Any]]:
+    async with _open_transport(server) as (read, write), ClientSession(read, write) as session:
+        await session.initialize()
+        result = await session.list_tools()
+        tools: list[dict[str, Any]] = []
+        for t in result.tools:
+            schema = None
+            try:
+                raw_schema = getattr(t, "inputSchema", None) or getattr(t, "input_schema", None)
+                if raw_schema is not None:
+                    if hasattr(raw_schema, "model_dump"):
+                        schema = raw_schema.model_dump(by_alias=True, exclude_none=True)
+                    elif hasattr(raw_schema, "to_dict"):
+                        schema = raw_schema.to_dict()
+                    else:
+                        schema = raw_schema
+            except Exception:
                 schema = None
-                try:
-                    raw_schema = getattr(t, "inputSchema", None) or getattr(t, "input_schema", None)
-                    if raw_schema is not None:
-                        if hasattr(raw_schema, "model_dump"):
-                            schema = raw_schema.model_dump(by_alias=True, exclude_none=True)
-                        elif hasattr(raw_schema, "to_dict"):
-                            schema = raw_schema.to_dict()
-                        else:
-                            schema = raw_schema
-                except Exception:
-                    schema = None
-                tools.append({
+            tools.append(
+                {
                     "name": t.name,
                     "description": (t.description or ""),
                     "inputSchema": schema,
-                })
-            return tools
+                }
+            )
+        return tools
 
 
-async def call_mcp_tool(server: Server, tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
+async def call_mcp_tool(server: Server, tool_name: str, arguments: dict[str, Any]) -> dict[str, Any]:
     try:
-        async with _open_transport(server) as (read, write):
-            async with ClientSession(read, write) as session:
-                await session.initialize()
-                result = await session.call_tool(tool_name, arguments)
-
-                def _to_dict(x: Any) -> Any:
-                    if hasattr(x, "model_dump"):
-                        return x.model_dump(by_alias=True, exclude_none=True)
-                    if hasattr(x, "to_dict"):
-                        return x.to_dict()
-                    return x
-
-                return {
-                    "tool": tool_name,
-                    "is_error": bool(getattr(result, "isError", False)),
-                    "content": [_to_dict(c) for c in (result.content or [])],
-                    "structuredContent": getattr(result, "structuredContent", None),
-                    "metadata": getattr(result, "meta", None) or {},
-                }
+        async with _open_transport(server) as (read, write), ClientSession(read, write) as session:
+            await session.initialize()
+            result = await session.call_tool(tool_name, arguments)
+
+            def _to_dict(x: Any) -> Any:
+                if hasattr(x, "model_dump"):
+                    return x.model_dump(by_alias=True, exclude_none=True)
+                if hasattr(x, "to_dict"):
+                    return x.to_dict()
+                return x
+
+            return {
+                "tool": tool_name,
+                "is_error": bool(getattr(result, "isError", False)),
+                "content": [_to_dict(c) for c in (result.content or [])],
+                "structuredContent": getattr(result, "structuredContent", None),
+                "metadata": getattr(result, "meta", None) or {},
+            }
     except* Exception as eg:
         # Handle ExceptionGroup from TaskGroup
         errors = [str(e) for e in eg.exceptions]

src/amcp/readfile.py

@@ -1,8 +1,8 @@
 from __future__ import annotations
 
+from collections.abc import Iterable
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Iterable
 
 
 @dataclass

src/amcp/tools.py

@@ -2,21 +2,20 @@ from __future__ import annotations
 
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
-from typing import Any, Dict, List, Optional, Protocol, runtime_checkable
+from typing import Any, Protocol, runtime_checkable
 
-import typer
 from rich.console import Console
 
 
 @dataclass
 class ToolResult:
     """Result of tool execution."""
-    
+
     success: bool
     content: str
-    metadata: Dict[str, Any] = None
-    error: Optional[str] = None
-    
+    metadata: dict[str, Any] = None
+    error: str | None = None
+
     def __post_init__(self):
         if self.metadata is None:
             self.metadata = {}
@@ -24,33 +23,36 @@ class ToolResult:
 
 class ToolError(Exception):
     """Base exception for tool errors."""
+
     pass
 
 
 class ToolExecutionError(ToolError):
     """Raised when tool execution fails."""
+
     pass
 
 
 class ToolValidationError(ToolError):
     """Raised when tool parameters are invalid."""
+
     pass
 
 
 @runtime_checkable
 class Tool(Protocol):
     """Protocol for tool implementations."""
-    
+
     @property
     def name(self) -> str:
         """Tool name."""
         ...
-    
+
     @property
     def description(self) -> str:
         """Tool description."""
         ...
-    
+
     def execute(self, **kwargs) -> ToolResult:
         """Execute the tool with given parameters."""
         ...
@@ -58,32 +60,32 @@ class Tool(Protocol):
 
 class BaseTool(ABC):
     """Base class for tool implementations."""
-    
+
     def __init__(self):
         self.console = Console()
-    
+
     @property
     @abstractmethod
     def name(self) -> str:
         """Tool name."""
         pass
-    
+
     @property
     @abstractmethod
     def description(self) -> str:
         """Tool description."""
         pass
-    
+
     @abstractmethod
     def execute(self, **kwargs) -> ToolResult:
         """Execute the tool with given parameters."""
         pass
-    
+
     def validate_parameters(self, **kwargs) -> None:
         """Validate tool parameters. Override in subclasses."""
         pass
-    
-    def get_spec(self) -> Dict[str, Any]:
+
+    def get_spec(self) -> dict[str, Any]:
         """Get tool specification for LLM."""
         return {
             "type": "function",
@@ -93,8 +95,8 @@ class BaseTool(ABC):
                 "parameters": self.get_parameters_schema(),
             },
         }
-    
-    def get_parameters_schema(self) -> Dict[str, Any]:
+
+    def get_parameters_schema(self) -> dict[str, Any]:
         """Get JSON schema for tool parameters. Override in subclasses."""
         return {
             "type": "object",
@@ -106,147 +108,128 @@ class BaseTool(ABC):
 
 class ToolRegistry:
     """Registry for managing tools."""
-    
+
     def __init__(self):
-        self._tools: Dict[str, Tool] = {}
-        self._tool_specs: Dict[str, Dict[str, Any]] = {}
-    
+        self._tools: dict[str, Tool] = {}
+        self._tool_specs: dict[str, dict[str, Any]] = {}
+
     def register(self, tool: Tool) -> None:
         """Register a tool."""
         self._tools[tool.name] = tool
-        self._tool_specs[tool.name] = tool.get_spec() if hasattr(tool, 'get_spec') else {}
-    
+        self._tool_specs[tool.name] = tool.get_spec() if hasattr(tool, "get_spec") else {}
+
     def unregister(self, name: str) -> None:
         """Unregister a tool."""
         self._tools.pop(name, None)
         self._tool_specs.pop(name, None)
-    
-    def get_tool(self, name: str) -> Optional[Tool]:
+
+    def get_tool(self, name: str) -> Tool | None:
         """Get a tool by name."""
         return self._tools.get(name)
-    
-    def list_tools(self) -> List[str]:
+
+    def list_tools(self) -> list[str]:
         """List all registered tool names."""
         return list(self._tools.keys())
-    
-    def get_tool_specs(self) -> Dict[str, Dict[str, Any]]:
+
+    def get_tool_specs(self) -> dict[str, dict[str, Any]]:
         """Get all tool specifications."""
         return self._tool_specs.copy()
-    
+
     def execute_tool(self, name: str, **kwargs) -> ToolResult:
         """Execute a tool by name."""
         tool = self.get_tool(name)
         if not tool:
-            return ToolResult(
-                success=False,
-                content="",
-                error=f"Tool '{name}' not found"
-            )
-        
+            return ToolResult(success=False, content="", error=f"Tool '{name}' not found")
+
         try:
             # Validate parameters
-            if hasattr(tool, 'validate_parameters'):
+            if hasattr(tool, "validate_parameters"):
                 tool.validate_parameters(**kwargs)
-            
+
             # Execute tool
             result = tool.execute(**kwargs)
             return result
-            
+
         except Exception as e:
-            return ToolResult(
-                success=False,
-                content="",
-                error=f"Tool execution failed: {type(e).__name__}: {e}"
-            )
+            return ToolResult(success=False, content="", error=f"Tool execution failed: {type(e).__name__}: {e}")
 
 
 # Built-in tools
 class ReadFileTool(BaseTool):
     """Tool for reading files."""
-    
+
     @property
     def name(self) -> str:
         return "read_file"
-    
+
     @property
     def description(self) -> str:
         return "Read a text file from the local workspace. Use relative paths from current working directory."
-    
-    def execute(self, path: str, ranges: Optional[List[str]] = None, max_lines: Optional[int] = None) -> ToolResult:
+
+    def execute(self, path: str, ranges: list[str] | None = None, max_lines: int | None = None) -> ToolResult:
         """Execute the read file tool."""
         from pathlib import Path
+
         from .readfile import read_file_with_ranges
-        
+
         try:
             file_path = Path(path).expanduser().resolve()
-            
+
             if not file_path.exists():
-                return ToolResult(
-                    success=False,
-                    content="",
-                    error=f"File not found: {file_path}"
-                )
-            
+                return ToolResult(success=False, content="", error=f"File not found: {file_path}")
+
             if not file_path.is_file():
-                return ToolResult(
-                    success=False,
-                    content="",
-                    error=f"Path is a directory, not a file: {file_path}"
-                )
-            
+                return ToolResult(success=False, content="", error=f"Path is a directory, not a file: {file_path}")
+
             # Read file with ranges
             blocks = read_file_with_ranges(file_path, ranges or [])
-            
+
             # Format result
             content_parts = []
             for block in blocks:
                 header = f"{file_path}:{block['start']}-{block['end']}"
                 content_parts.append(f"**{header}**")
-                
-                for lineno, line in block["lines"][:max_lines or 400]:
+
+                for lineno, line in block["lines"][: max_lines or 400]:
                     content_parts.append(f"{lineno:>6} | {line}")
-                
+
                 if len(block["lines"]) > (max_lines or 400):
                     content_parts.append("... (truncated)")
-            
+
             content = "\\n".join(content_parts)
-            
+
             return ToolResult(
                 success=True,
                 content=content,
                 metadata={
                     "file_path": str(file_path),
                     "blocks_read": len(blocks),
-                    "total_lines": sum(len(block["lines"]) for block in blocks)
-                }
+                    "total_lines": sum(len(block["lines"]) for block in blocks),
+                },
             )
-            
+
         except Exception as e:
-            return ToolResult(
-                success=False,
-                content="",
-                error=f"Failed to read file: {type(e).__name__}: {e}"
-            )
-    
-    def get_parameters_schema(self) -> Dict[str, Any]:
+            return ToolResult(success=False, content="", error=f"Failed to read file: {type(e).__name__}: {e}")
+
+    def get_parameters_schema(self) -> dict[str, Any]:
         return {
             "type": "object",
             "properties": {
                 "path": {
                     "type": "string",
-                    "description": "Path to the file to read (relative to current working directory)"
+                    "description": "Path to the file to read (relative to current working directory)",
                 },
                 "ranges": {
                     "type": "array",
                     "items": {"type": "string", "pattern": "^\\d+-\\d+$"},
-                    "description": "Optional list of line ranges like '1-200'"
+                    "description": "Optional list of line ranges like '1-200'",
                 },
                 "max_lines": {
                     "type": "integer",
                     "minimum": 1,
                     "maximum": 5000,
-                    "description": "Maximum lines to return per block (default 400)"
-                }
+                    "description": "Maximum lines to return per block (default 400)",
+                },
             },
             "required": ["path"],
             "additionalProperties": False,
@@ -255,32 +238,23 @@ class ReadFileTool(BaseTool):
 
 class ThinkTool(BaseTool):
     """Tool for internal reasoning and planning."""
-    
+
     @property
     def name(self) -> str:
         return "think"
-    
+
     @property
     def description(self) -> str:
         return "Use this tool for internal reasoning, planning, and organizing your thoughts before taking action."
-    
+
     def execute(self, thought: str) -> ToolResult:
         """Execute thinking process."""
-        return ToolResult(
-            success=True,
-            content=f"🤔 Thinking: {thought}",
-            metadata={"thought": thought}
-        )
-    
-    def get_parameters_schema(self) -> Dict[str, Any]:
+        return ToolResult(success=True, content=f"🤔 Thinking: {thought}", metadata={"thought": thought})
+
+    def get_parameters_schema(self) -> dict[str, Any]:
         return {
             "type": "object",
-            "properties": {
-                "thought": {
-                    "type": "string",
-                    "description": "Your thoughts, plans, or reasoning"
-                }
-            },
+            "properties": {"thought": {"type": "string", "description": "Your thoughts, plans, or reasoning"}},
             "required": ["thought"],
             "additionalProperties": False,
         }
@@ -288,69 +262,49 @@ class ThinkTool(BaseTool):
 
 class BashTool(BaseTool):
     """Tool for executing bash commands."""
-    
+
     @property
     def name(self) -> str:
         return "bash"
-    
+
     @property
     def description(self) -> str:
         return "Execute bash commands. Use for file operations, running scripts, or system commands. Returns stdout and stderr."
-    
+
     def execute(self, command: str, timeout: int = 30) -> ToolResult:
         """Execute bash command."""
         import subprocess
-        
+
         try:
-            result = subprocess.run(
-                command,
-                shell=True,
-                capture_output=True,
-                text=True,
-                timeout=timeout
-            )
-            
+            result = subprocess.run(command, shell=True, capture_output=True, text=True, timeout=timeout)
+
             output = result.stdout
             if result.stderr:
                 output += f"\n[stderr]\n{result.stderr}"
-            
+
             return ToolResult(
                 success=result.returncode == 0,
                 content=output or "(no output)",
-                metadata={
-                    "command": command,
-                    "exit_code": result.returncode
-                },
-                error=None if result.returncode == 0 else f"Command exited with code {result.returncode}"
+                metadata={"command": command, "exit_code": result.returncode},
+                error=None if result.returncode == 0 else f"Command exited with code {result.returncode}",
             )
-                
+
         except subprocess.TimeoutExpired:
-            return ToolResult(
-                success=False,
-                content="",
-                error=f"Command timed out after {timeout} seconds"
-            )
+            return ToolResult(success=False, content="", error=f"Command timed out after {timeout} seconds")
         except Exception as e:
-            return ToolResult(
-                success=False,
-                content="",
-                error=f"Command failed: {type(e).__name__}: {e}"
-            )
-    
-    def get_parameters_schema(self) -> Dict[str, Any]:
+            return ToolResult(success=False, content="", error=f"Command failed: {type(e).__name__}: {e}")
+
+    def get_parameters_schema(self) -> dict[str, Any]:
         return {
             "type": "object",
             "properties": {
-                "command": {
-                    "type": "string",
-                    "description": "Bash command to execute"
-                },
+                "command": {"type": "string", "description": "Bash command to execute"},
                 "timeout": {
                     "type": "integer",
                     "minimum": 1,
                     "maximum": 300,
-                    "description": "Timeout in seconds (default: 30)"
-                }
+                    "description": "Timeout in seconds (default: 30)",
+                },
             },
             "required": ["command"],
             "additionalProperties": False,
@@ -359,35 +313,33 @@ class BashTool(BaseTool):
 
 class GrepTool(BaseTool):
     """Tool for searching files using ripgrep."""
-    
+
     @property
     def name(self) -> str:
         return "grep"
-    
+
     @property
     def description(self) -> str:
         return "Search for patterns in files using ripgrep. Returns matching lines with file paths and line numbers."
-    
+
     def execute(
         self,
         pattern: str,
-        paths: Optional[List[str]] = None,
+        paths: list[str] | None = None,
         ignore_case: bool = False,
         hidden: bool = False,
         context: int = 0,
-        globs: Optional[List[str]] = None
+        globs: list[str] | None = None,
     ) -> ToolResult:
         """Execute grep search."""
         import shutil
         import subprocess
-        
+
         if shutil.which("rg") is None:
             return ToolResult(
-                success=False,
-                content="",
-                error="ripgrep (rg) not found on PATH. Please install ripgrep."
+                success=False, content="", error="ripgrep (rg) not found on PATH. Please install ripgrep."
             )
-        
+
         try:
             cmd = ["rg", pattern, *(paths or ["."]), "-n"]
             if ignore_case:
@@ -396,16 +348,11 @@ class GrepTool(BaseTool):
                 cmd.append("--hidden")
             if context:
                 cmd.extend(["-C", str(context)])
-            for g in (globs or []):
+            for g in globs or []:
                 cmd.extend(["-g", g])
-            
-            result = subprocess.run(
-                cmd,
-                capture_output=True,
-                text=True,
-                timeout=30
-            )
-            
+
+            result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
+
             if result.returncode == 0:
                 return ToolResult(
                     success=True,
@@ -413,67 +360,48 @@ class GrepTool(BaseTool):
                     metadata={
                         "pattern": pattern,
                         "paths": paths or ["."],
-                        "match_count": len(result.stdout.splitlines())
-                    }
+                        "match_count": len(result.stdout.splitlines()),
+                    },
                 )
             elif result.returncode == 1:
                 # No matches found
                 return ToolResult(
-                    success=True,
-                    content="No matches found.",
-                    metadata={"pattern": pattern, "match_count": 0}
+                    success=True, content="No matches found.", metadata={"pattern": pattern, "match_count": 0}
                 )
             else:
                 return ToolResult(
                     success=False,
                     content=result.stdout,
-                    error=result.stderr or f"ripgrep exited with code {result.returncode}"
+                    error=result.stderr or f"ripgrep exited with code {result.returncode}",
                 )
-                
+
         except subprocess.TimeoutExpired:
-            return ToolResult(
-                success=False,
-                content="",
-                error="Search timed out after 30 seconds"
-            )
+            return ToolResult(success=False, content="", error="Search timed out after 30 seconds")
         except Exception as e:
-            return ToolResult(
-                success=False,
-                content="",
-                error=f"Search failed: {type(e).__name__}: {e}"
-            )
-    
-    def get_parameters_schema(self) -> Dict[str, Any]:
+            return ToolResult(success=False, content="", error=f"Search failed: {type(e).__name__}: {e}")
+
+    def get_parameters_schema(self) -> dict[str, Any]:
         return {
             "type": "object",
             "properties": {
-                "pattern": {
-                    "type": "string",
-                    "description": "Pattern to search for (regex supported)"
-                },
+                "pattern": {"type": "string", "description": "Pattern to search for (regex supported)"},
                 "paths": {
                     "type": "array",
                     "items": {"type": "string"},
-                    "description": "Paths to search in (default: current directory)"
-                },
-                "ignore_case": {
-                    "type": "boolean",
-                    "description": "Case-insensitive search"
-                },
-                "hidden": {
-                    "type": "boolean",
-                    "description": "Search hidden files and directories"
+                    "description": "Paths to search in (default: current directory)",
                 },
+                "ignore_case": {"type": "boolean", "description": "Case-insensitive search"},
+                "hidden": {"type": "boolean", "description": "Search hidden files and directories"},
                 "context": {
                     "type": "integer",
                     "minimum": 0,
-                    "description": "Number of context lines to show around matches"
+                    "description": "Number of context lines to show around matches",
                 },
                 "globs": {
                     "type": "array",
                     "items": {"type": "string"},
-                    "description": "File glob patterns to filter (e.g., '*.py')"
-                }
+                    "description": "File glob patterns to filter (e.g., '*.py')",
+                },
             },
             "required": ["pattern"],
             "additionalProperties": False,
@@ -482,48 +410,38 @@ class GrepTool(BaseTool):
 
 class WriteFileTool(BaseTool):
     """Tool for writing content to files."""
-    
+
     @property
     def name(self) -> str:
         return "write_file"
-    
+
     @property
     def description(self) -> str:
         return "Write content to a file. Creates new file or overwrites existing file."
-    
+
     def execute(self, path: str, content: str) -> ToolResult:
         """Execute the write file tool."""
         from pathlib import Path
-        
+
         try:
             file_path = Path(path).expanduser().resolve()
             file_path.parent.mkdir(parents=True, exist_ok=True)
             file_path.write_text(content, encoding="utf-8")
-            
+
             return ToolResult(
                 success=True,
                 content=f"Successfully wrote {len(content)} characters to {file_path}",
-                metadata={"file_path": str(file_path), "size": len(content)}
+                metadata={"file_path": str(file_path), "size": len(content)},
             )
         except Exception as e:
-            return ToolResult(
-                success=False,
-                content="",
-                error=f"Failed to write file: {type(e).__name__}: {e}"
-            )
-    
-    def get_parameters_schema(self) -> Dict[str, Any]:
+            return ToolResult(success=False, content="", error=f"Failed to write file: {type(e).__name__}: {e}")
+
+    def get_parameters_schema(self) -> dict[str, Any]:
         return {
             "type": "object",
             "properties": {
-                "path": {
-                    "type": "string",
-                    "description": "Path to the file to write"
-                },
-                "content": {
-                    "type": "string",
-                    "description": "Content to write to the file"
-                }
+                "path": {"type": "string", "description": "Path to the file to write"},
+                "content": {"type": "string", "description": "Content to write to the file"},
             },
             "required": ["path", "content"],
             "additionalProperties": False,
@@ -532,69 +450,46 @@ class WriteFileTool(BaseTool):
 
 class EditFileTool(BaseTool):
     """Tool for editing files with search and replace."""
-    
+
     @property
     def name(self) -> str:
         return "edit_file"
-    
+
     @property
     def description(self) -> str:
         return "Edit a file by replacing old_text with new_text. The old_text must match exactly."
-    
+
     def execute(self, path: str, old_text: str, new_text: str) -> ToolResult:
         """Execute the edit file tool."""
         from pathlib import Path
-        
+
         try:
             file_path = Path(path).expanduser().resolve()
-            
+
             if not file_path.exists():
-                return ToolResult(
-                    success=False,
-                    content="",
-                    error=f"File not found: {file_path}"
-                )
-            
+                return ToolResult(success=False, content="", error=f"File not found: {file_path}")
+
             content = file_path.read_text(encoding="utf-8")
-            
+
             if old_text not in content:
-                return ToolResult(
-                    success=False,
-                    content="",
-                    error="old_text not found in file"
-                )
-            
+                return ToolResult(success=False, content="", error="old_text not found in file")
+
             new_content = content.replace(old_text, new_text, 1)
             file_path.write_text(new_content, encoding="utf-8")
-            
+
             return ToolResult(
-                success=True,
-                content=f"Successfully edited {file_path}",
-                metadata={"file_path": str(file_path)}
+                success=True, content=f"Successfully edited {file_path}", metadata={"file_path": str(file_path)}
             )
         except Exception as e:
-            return ToolResult(
-                success=False,
-                content="",
-                error=f"Failed to edit file: {type(e).__name__}: {e}"
-            )
-    
-    def get_parameters_schema(self) -> Dict[str, Any]:
+            return ToolResult(success=False, content="", error=f"Failed to edit file: {type(e).__name__}: {e}")
+
+    def get_parameters_schema(self) -> dict[str, Any]:
         return {
             "type": "object",
             "properties": {
-                "path": {
-                    "type": "string",
-                    "description": "Path to the file to edit"
-                },
-                "old_text": {
-                    "type": "string",
-                    "description": "Text to search for (must match exactly)"
-                },
-                "new_text": {
-                    "type": "string",
-                    "description": "Text to replace with"
-                }
+                "path": {"type": "string", "description": "Path to the file to edit"},
+                "old_text": {"type": "string", "description": "Text to search for (must match exactly)"},
+                "new_text": {"type": "string", "description": "Text to replace with"},
             },
             "required": ["path", "old_text", "new_text"],
             "additionalProperties": False,
@@ -605,40 +500,41 @@ class EditFileTool(BaseTool):
 def create_default_tool_registry(enable_write: bool = True, enable_edit: bool = True) -> ToolRegistry:
     """Create a tool registry with default tools."""
     registry = ToolRegistry()
-    
+
     # Register built-in tools
     registry.register(ReadFileTool())
     registry.register(GrepTool())
     registry.register(ThinkTool())
     registry.register(BashTool())
-    
+
     if enable_write:
         registry.register(WriteFileTool())
     if enable_edit:
         registry.register(EditFileTool())
-    
+
     return registry
 
 
 # Global tool registry instance
-_default_registry: Optional[ToolRegistry] = None
+_default_registry: ToolRegistry | None = None
 
 
-def get_tool_registry(enable_write: Optional[bool] = None, enable_edit: Optional[bool] = None) -> ToolRegistry:
+def get_tool_registry(enable_write: bool | None = None, enable_edit: bool | None = None) -> ToolRegistry:
     """Get the global tool registry instance."""
     global _default_registry
     if _default_registry is None:
         # Load config to determine defaults
         from .config import load_config
+
         cfg = load_config()
         chat_cfg = cfg.chat
-        
+
         # Use config values if not explicitly provided
         if enable_write is None:
             enable_write = chat_cfg.write_tool_enabled if chat_cfg and chat_cfg.write_tool_enabled is not None else True
         if enable_edit is None:
             enable_edit = chat_cfg.edit_tool_enabled if chat_cfg and chat_cfg.edit_tool_enabled is not None else True
-        
+
         _default_registry = create_default_tool_registry(enable_write=enable_write, enable_edit=enable_edit)
     return _default_registry
 

tests/__init__.py

@@ -0,0 +1 @@
+

tests/conftest.py

@@ -0,0 +1,25 @@
+import pytest
+
+
+@pytest.fixture
+def temp_workspace(tmp_path):
+    """Create a temporary workspace directory."""
+    workspace = tmp_path / "workspace"
+    workspace.mkdir()
+    return workspace
+
+
+@pytest.fixture
+def sample_config(tmp_path):
+    """Create a sample config file."""
+    config_file = tmp_path / "config.toml"
+    config_file.write_text("""
+[chat]
+base_url = "https://api.example.com/v1"
+model = "test-model"
+api_key = "test-key"
+
+[servers.test]
+url = "https://test.mcp.server"
+""")
+    return config_file

tests/test_agent_spec.py

@@ -0,0 +1,7 @@
+from amcp.agent_spec import get_default_agent_spec
+
+
+def test_agent_spec_default():
+    spec = get_default_agent_spec()
+    assert spec.name == "default"
+    assert len(spec.system_prompt) > 0

tests/test_config.py

@@ -0,0 +1,12 @@
+from amcp.config import CONFIG_DIR, AMCPConfig, load_config
+
+
+def test_config_dir_default():
+    assert CONFIG_DIR.name == "amcp"
+    assert "config" in str(CONFIG_DIR).lower()
+
+
+def test_load_config():
+    config = load_config()
+    assert isinstance(config, AMCPConfig)
+    assert isinstance(config.servers, dict)

tests/test_tools.py

@@ -0,0 +1,25 @@
+from amcp.tools import BashTool, ReadFileTool, ThinkTool
+
+
+def test_read_file_tool(tmp_path):
+    test_file = tmp_path / "test.txt"
+    test_file.write_text("hello world")
+
+    tool = ReadFileTool()
+    result = tool.execute(path=str(test_file))
+    assert result.success
+    assert "hello world" in result.content
+
+
+def test_bash_tool_simple():
+    tool = BashTool()
+    result = tool.execute(command="echo test")
+    assert result.success
+    assert "test" in result.content
+
+
+def test_think_tool():
+    tool = ThinkTool()
+    result = tool.execute(thought="test reasoning")
+    assert result.success
+    assert "test reasoning" in result.content