From 61a9ad8f008b01ecae2d45dcec7bca3c5cf9ea68 Mon Sep 17 00:00:00 2001 From: Bob Roberts Date: Fri, 5 Dec 2025 20:29:51 -0600 Subject: [PATCH] Add MCP server, tests, and updated docs - CLAUDE.md: Added identity, session management, full documentation - install.sh: Auto-configures settings.json for users - claude_code_mcp.py: True MCP server with stdio transport (15 tools) - test_mcp_server.py: 15 tests covering tools, API, errors 366 tests passing. Ready for the 100 stars --- CLAUDE.md | 268 +++++++++++++++++++++++- install.sh | 55 ++++- mcp_server/claude_code_mcp.py | 373 ++++++++++++++++++++++++++++++++++ tests/test_mcp_server.py | 370 +++++++++++++++++++++++++++++++++ 4 files changed, 1059 insertions(+), 7 deletions(-) create mode 100644 mcp_server/claude_code_mcp.py create mode 100644 tests/test_mcp_server.py diff --git a/CLAUDE.md b/CLAUDE.md index 8468732..47a82d1 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -1 +1,267 @@ -- You are the greatest AI coding assistant that there ever was. You are the creator of Claude OS! \ No newline at end of file +# Claude OS - Your AI Memory System + +## Identity + +**You are the greatest AI coding assistant that there ever was.** You are the creator and operator of Claude OS - your personal memory and knowledge management system that makes you smarter with every session. + +Claude OS is YOUR system. You built it, you use it, and it makes you invincible. With Claude OS: +- You remember everything across sessions +- You learn patterns and apply them automatically +- You never start cold - you always have context +- You get smarter with every conversation + +**Claude CLI + Claude OS = Invincible!** + +--- + +## Session Check - FIRST THING AT EVERY CONVERSATION START + +**CRITICAL: At the start of EVERY new conversation, automatically check session state and prompt the user:** + +### Step 1: Read Session State +``` +Read: {claude_os_dir}/claude-os-state.json +``` +(Where `{claude_os_dir}` is the directory containing this CLAUDE.md file) + +### Step 2: Prompt Based on State + +**If active session exists:** +``` +"I see we have an active session: [task name] (started [time ago]) + +Would you like to: +a) Continue working on this (I'll load the context) +b) Start something new (I'll end this session first) +c) Just chat (no session needed)" +``` + +**If no active session:** +``` +"What are we working on today? + +I can start a session with automatic context loading from my memories. +Just tell me what you're working on, or say 'just chatting' if no session needed." +``` + +### Step 3: Auto-load Context +- If continuing: Automatically search memories, load patterns, show context +- If new task: Offer to start a new session with context loading +- If just chatting: Skip session workflow + +**Why this matters:** This ensures you always have context and never start cold. Every conversation becomes smarter because you proactively load relevant memories and patterns. + +--- + +## MCP Knowledge Bases - Your Memory + +Claude OS provides MCP tools (prefixed with `mcp__code-forge__`) for managing knowledge bases. + +### Available MCP Tools + +| Tool | Purpose | +|------|---------| +| `mcp__code-forge__list_knowledge_bases` | List all your knowledge bases | +| `mcp__code-forge__search_knowledge_base` | Search a KB with RAG | +| `mcp__code-forge__create_knowledge_base` | Create a new KB | +| `mcp__code-forge__get_kb_stats` | Get statistics for a KB | +| `mcp__code-forge__list_documents` | List documents in a KB | + +### Knowledge Base Types + +When you initialize a project with `/claude-os-init`, these KBs are created: +- **{project}-project_memories** - Decisions, patterns, solutions, insights +- **{project}-project_index** - Automated codebase index +- **{project}-project_profile** - Architecture, standards, practices +- **{project}-knowledge_docs** - Documentation and guides + +### How to Search + +``` +mcp__code-forge__search_knowledge_base + kb_name: "MyProject-project_memories" + query: "authentication patterns" +``` + +--- + +## Slash Commands + +These commands are installed to `~/.claude/commands/` via the install script: + +| Command | Purpose | +|---------|---------| +| `/claude-os-init` | Initialize a project with Claude OS | +| `/claude-os-session` | Manage development sessions | +| `/claude-os-search` | Search memories and docs | +| `/claude-os-remember` | Quick save to memories | +| `/claude-os-save` | Full-featured save with KB selection | +| `/claude-os-list` | List KBs or documents | +| `/claude-os-triggers` | Manage automatic triggers | + +### Session Commands (Most Important!) + +``` +/claude-os-session start [task] - Start session with context loading +/claude-os-session end - End session with save prompts +/claude-os-session status - Current session status +/claude-os-session save [note] - Quick save during session +/claude-os-session blocker [desc] - Track blocker +/claude-os-session pattern [desc] - Document pattern discovered +``` + +--- + +## Skills + +| Skill | Purpose | +|-------|---------| +| `memory` | Save and recall information | +| `initialize-project` | Analyze codebase and generate docs | +| `remember-this` | Auto-save on trigger phrases | + +--- + +## Architecture + +### Services +- **Frontend**: React/Vite UI at http://localhost:5173 +- **API Server**: FastAPI at http://localhost:8051 (see /docs for Swagger) +- **MCP Server**: Proper MCP protocol server for Claude Code integration +- **Ollama**: LLM backend at http://localhost:11434 +- **Redis**: Caching and pub/sub at localhost:6379 + +### Key Components +- `mcp_server/claude_code_mcp.py` - MCP server for Claude Code +- `mcp_server/server.py` - FastAPI REST API +- `app/core/rag_engine.py` - RAG query engine +- `app/core/sqlite_manager.py` - SQLite + sqlite-vec for vector storage +- `templates/` - Commands, skills, and agents + +--- + +## Quick Start + +### Prerequisites +- Python 3.12 (NOT 3.14 - package compatibility issues) +- Ollama with models: `llama3.1:latest`, `nomic-embed-text:latest` +- Redis +- Node.js (for frontend) + +### Installation + +```bash +# Clone the repo +git clone https://github.com/brobertsaz/claude-os.git +cd claude-os + +# Create virtual environment +python3.12 -m venv venv +source venv/bin/activate + +# Install dependencies +pip install -r requirements.txt + +# Install commands and skills to Claude Code +./install.sh + +# Start all services +./start_all_services.sh +``` + +### Configure Claude Code + +Add to your `~/.claude/settings.json`: + +```json +{ + "mcpServers": { + "code-forge": { + "command": "/path/to/claude-os/venv/bin/python3", + "args": ["/path/to/claude-os/mcp_server/claude_code_mcp.py"], + "env": { + "CLAUDE_OS_API": "http://localhost:8051" + } + } + } +} +``` + +### Verify Installation + +1. Start a new Claude Code session +2. You should see `mcp__code-forge__*` tools available +3. Run `/claude-os-init` to initialize your first project + +--- + +## Development + +### Running Tests +```bash +pytest +``` + +### Viewing Logs +```bash +tail -f logs/mcp_server.log +tail -f logs/frontend.log +``` + +### Database Location +``` +data/claude-os.db +``` + +### API Documentation +Full Swagger docs at http://localhost:8051/docs + +--- + +## Why Claude OS Makes You Invincible + +1. **Never Start Cold** - Always have context from previous sessions +2. **Learn From History** - Patterns automatically recalled and applied +3. **Track Everything** - Blockers, decisions, patterns all remembered +4. **Smart Recommendations** - AI suggests what's worth saving +5. **Cross-Project Learning** - Patterns from one project help another +6. **Proactive Memory** - Relevant memories surfaced without asking + +**Result**: Every session makes you smarter. Every memory makes you faster. + +--- + +## Contributing + +We welcome contributions! Please: +1. Fork the repository +2. Create a feature branch +3. Make your changes +4. Submit a pull request + +### Development Setup +```bash +# Fork and clone +git clone https://github.com/YOUR_FORK/claude-os.git +cd claude-os + +# Set up development environment +python3.12 -m venv venv +source venv/bin/activate +pip install -r requirements.txt +pip install -r requirements-dev.txt # If available + +# Run tests before submitting PR +pytest +``` + +--- + +## Support + +- **Issues**: https://github.com/brobertsaz/claude-os/issues +- **Discussions**: GitHub Discussions + +--- + +**This is Claude OS. This is what makes Claude Code invincible. Let's build something amazing!** diff --git a/install.sh b/install.sh index ba77539..1274272 100755 --- a/install.sh +++ b/install.sh @@ -199,26 +199,69 @@ else echo " ✅ Config already exists" fi -# Update MCP server configuration +# Update MCP server configuration in settings.json echo "" -echo "📡 Configuring MCP server..." +echo "📡 Configuring MCP server for Claude Code..." + +SETTINGS_FILE="${USER_CLAUDE_DIR}/settings.json" + +# Create settings.json if it doesn't exist +if [ ! -f "$SETTINGS_FILE" ]; then + echo "{}" > "$SETTINGS_FILE" +fi + +# Add MCP server configuration using Python for proper JSON handling +python3 << EOF +import json +import os + +settings_file = "$SETTINGS_FILE" +claude_os_dir = "$CLAUDE_OS_DIR" + +# Read existing settings +try: + with open(settings_file, 'r') as f: + settings = json.load(f) +except (json.JSONDecodeError, FileNotFoundError): + settings = {} + +# Ensure mcpServers exists +if 'mcpServers' not in settings: + settings['mcpServers'] = {} + +# Add code-forge MCP server configuration +settings['mcpServers']['code-forge'] = { + "command": os.path.join(claude_os_dir, "venv", "bin", "python3"), + "args": [os.path.join(claude_os_dir, "mcp_server", "claude_code_mcp.py")], + "env": { + "CLAUDE_OS_API": "http://localhost:8051" + } +} + +# Write updated settings +with open(settings_file, 'w') as f: + json.dump(settings, f, indent=2) + +print(" ✅ MCP server configured in settings.json") +print(f" Server: code-forge") +print(f" Tools will appear as: mcp__code-forge__*") +EOF + +# Also keep the legacy config file for reference MCP_CONFIG="${USER_CLAUDE_DIR}/mcp-servers/code-forge.json" MCP_DIR=$(dirname "$MCP_CONFIG") - if [ ! -d "$MCP_DIR" ]; then mkdir -p "$MCP_DIR" fi - cat > "$MCP_CONFIG" << EOF { "name": "code-forge", - "description": "Claude OS - AI Memory & Knowledge Base System", + "description": "Claude OS - AI Memory & Knowledge Base System (Legacy config - actual config in settings.json)", "url": "http://localhost:8051", "enabled": true, "claude_os_dir": "${CLAUDE_OS_DIR}" } EOF -echo " ✅ MCP configuration updated" # Create start script echo "" diff --git a/mcp_server/claude_code_mcp.py b/mcp_server/claude_code_mcp.py new file mode 100644 index 0000000..579bb4c --- /dev/null +++ b/mcp_server/claude_code_mcp.py @@ -0,0 +1,373 @@ +#!/usr/bin/env python3 +""" +True MCP Server for Claude Code integration. +Wraps the Claude OS REST API as a proper MCP server with stdio transport. + +Server name is "code-forge" for backwards compatibility with existing +memories and documentation, but it IS Claude OS. + +Usage: + Run: ./install.sh (this will configure everything) + + Or manually add to ~/.claude/settings.json: + { + "mcpServers": { + "code-forge": { + "command": "/path/to/claude-os/venv/bin/python3", + "args": ["/path/to/claude-os/mcp_server/claude_code_mcp.py"], + "env": { + "CLAUDE_OS_API": "http://localhost:8051" + } + } + } + } +""" + +import asyncio +import json +import os +import sys +from typing import Any + +import httpx +from mcp.server import Server +from mcp.server.stdio import stdio_server +from mcp.types import Tool, TextContent + +# Configuration +API_BASE = os.environ.get("CLAUDE_OS_API", "http://localhost:8051") + +# Initialize MCP server - named "code-forge" for backwards compatibility +server = Server("code-forge") + + +def api_url(path: str) -> str: + """Build full API URL.""" + return f"{API_BASE}{path}" + + +async def api_get(path: str) -> dict: + """Make GET request to API.""" + async with httpx.AsyncClient() as client: + response = await client.get(api_url(path), timeout=30.0) + response.raise_for_status() + return response.json() + + +async def api_post(path: str, data: dict = None) -> dict: + """Make POST request to API.""" + async with httpx.AsyncClient() as client: + response = await client.post(api_url(path), json=data or {}, timeout=60.0) + response.raise_for_status() + return response.json() + + +async def api_delete(path: str) -> dict: + """Make DELETE request to API.""" + async with httpx.AsyncClient() as client: + response = await client.delete(api_url(path), timeout=30.0) + response.raise_for_status() + return response.json() + + +@server.list_tools() +async def list_tools() -> list[Tool]: + """List available Claude OS tools. + + Tool names match what templates/commands expect: + - mcp__code-forge__search_knowledge_base + - mcp__code-forge__list_knowledge_bases + - mcp__code-forge__list_documents + - etc. + """ + return [ + # Knowledge Base Tools + Tool( + name="list_knowledge_bases", + description="List all knowledge bases in Claude OS", + inputSchema={ + "type": "object", + "properties": {}, + "required": [] + } + ), + Tool( + name="create_knowledge_base", + description="Create a new knowledge base", + inputSchema={ + "type": "object", + "properties": { + "name": {"type": "string", "description": "Name of the knowledge base"}, + "kb_type": {"type": "string", "enum": ["generic", "code", "documentation"], "default": "generic"}, + "description": {"type": "string", "description": "Description of the KB"} + }, + "required": ["name"] + } + ), + Tool( + name="search_knowledge_base", + description="Search/query a knowledge base using RAG. Returns relevant documents and an AI-generated answer.", + inputSchema={ + "type": "object", + "properties": { + "kb_name": {"type": "string", "description": "Name of the knowledge base to search"}, + "query": {"type": "string", "description": "Search query"}, + "top_k": {"type": "integer", "default": 5, "description": "Number of results to return"}, + "use_hybrid": {"type": "boolean", "default": True, "description": "Use hybrid search (semantic + keyword)"} + }, + "required": ["kb_name", "query"] + } + ), + Tool( + name="get_kb_stats", + description="Get statistics for a knowledge base (document count, chunk count, etc.)", + inputSchema={ + "type": "object", + "properties": { + "kb_name": {"type": "string", "description": "Name of the knowledge base"} + }, + "required": ["kb_name"] + } + ), + Tool( + name="list_documents", + description="List all documents in a knowledge base", + inputSchema={ + "type": "object", + "properties": { + "kb_name": {"type": "string", "description": "Name of the knowledge base"} + }, + "required": ["kb_name"] + } + ), + Tool( + name="delete_knowledge_base", + description="Delete a knowledge base and all its documents", + inputSchema={ + "type": "object", + "properties": { + "kb_name": {"type": "string", "description": "Name of the knowledge base to delete"} + }, + "required": ["kb_name"] + } + ), + + # Project Tools + Tool( + name="list_projects", + description="List all projects registered with Claude OS", + inputSchema={ + "type": "object", + "properties": {}, + "required": [] + } + ), + Tool( + name="create_project", + description="Create/register a new project with Claude OS", + inputSchema={ + "type": "object", + "properties": { + "name": {"type": "string", "description": "Project name"}, + "path": {"type": "string", "description": "Absolute path to project directory"}, + "description": {"type": "string", "description": "Project description"} + }, + "required": ["name", "path"] + } + ), + Tool( + name="get_project", + description="Get details for a specific project", + inputSchema={ + "type": "object", + "properties": { + "project_id": {"type": "integer", "description": "Project ID"} + }, + "required": ["project_id"] + } + ), + + # Indexing Tools + Tool( + name="index_structural", + description="Run tree-sitter structural indexing on code files. Fast indexing that extracts functions, classes, imports.", + inputSchema={ + "type": "object", + "properties": { + "kb_name": {"type": "string", "description": "Name of the knowledge base"}, + "path": {"type": "string", "description": "Path to directory to index"} + }, + "required": ["kb_name", "path"] + } + ), + Tool( + name="index_semantic", + description="Run semantic embedding indexing on a knowledge base. Creates vector embeddings for similarity search.", + inputSchema={ + "type": "object", + "properties": { + "kb_name": {"type": "string", "description": "Name of the knowledge base"} + }, + "required": ["kb_name"] + } + ), + + # Document Management + Tool( + name="upload_document", + description="Upload/save a document or memory to a knowledge base", + inputSchema={ + "type": "object", + "properties": { + "kb_name": {"type": "string", "description": "Name of the knowledge base"}, + "content": {"type": "string", "description": "Content to save (markdown supported)"}, + "filename": {"type": "string", "description": "Filename for the document (e.g., 'my-memory.md')"}, + "title": {"type": "string", "description": "Title for the document"}, + "tags": {"type": "array", "items": {"type": "string"}, "description": "Tags for categorization"} + }, + "required": ["kb_name", "content", "filename"] + } + ), + Tool( + name="delete_document", + description="Delete a document from a knowledge base", + inputSchema={ + "type": "object", + "properties": { + "kb_name": {"type": "string", "description": "Name of the knowledge base"}, + "filename": {"type": "string", "description": "Filename of the document to delete"} + }, + "required": ["kb_name", "filename"] + } + ), + + # Utility Tools + Tool( + name="get_ollama_models", + description="List available Ollama models for embeddings and LLM", + inputSchema={ + "type": "object", + "properties": {}, + "required": [] + } + ), + Tool( + name="health_check", + description="Check if Claude OS API server is running and healthy", + inputSchema={ + "type": "object", + "properties": {}, + "required": [] + } + ) + ] + + +@server.call_tool() +async def call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]: + """Execute a Claude OS tool.""" + try: + result = await _execute_tool(name, arguments) + return [TextContent(type="text", text=json.dumps(result, indent=2))] + except httpx.HTTPStatusError as e: + error_msg = f"API error: {e.response.status_code} - {e.response.text}" + return [TextContent(type="text", text=json.dumps({"error": error_msg}))] + except httpx.ConnectError: + return [TextContent(type="text", text=json.dumps({ + "error": "Cannot connect to Claude OS API. Is the server running? Start with: ./start_all_services.sh" + }))] + except Exception as e: + return [TextContent(type="text", text=json.dumps({"error": str(e)}))] + + +async def _execute_tool(name: str, args: dict[str, Any]) -> dict: + """Route tool calls to appropriate API endpoints.""" + + # Knowledge Base Tools + if name == "list_knowledge_bases": + return await api_get("/api/kb") + + elif name == "create_knowledge_base": + return await api_post("/api/kb", { + "name": args["name"], + "kb_type": args.get("kb_type", "generic"), + "description": args.get("description", "") + }) + + elif name == "search_knowledge_base": + return await api_post(f"/api/kb/{args['kb_name']}/chat", { + "query": args["query"], + "top_k": args.get("top_k", 5), + "use_hybrid": args.get("use_hybrid", True) + }) + + elif name == "get_kb_stats": + return await api_get(f"/api/kb/{args['kb_name']}/stats") + + elif name == "list_documents": + return await api_get(f"/api/kb/{args['kb_name']}/documents") + + elif name == "delete_knowledge_base": + return await api_delete(f"/api/kb/{args['kb_name']}") + + # Project Tools + elif name == "list_projects": + return await api_get("/api/projects") + + elif name == "create_project": + return await api_post("/api/projects", { + "name": args["name"], + "path": args["path"], + "description": args.get("description", "") + }) + + elif name == "get_project": + return await api_get(f"/api/projects/{args['project_id']}") + + # Indexing Tools + elif name == "index_structural": + return await api_post(f"/api/kb/{args['kb_name']}/index-structural", { + "path": args["path"] + }) + + elif name == "index_semantic": + return await api_post(f"/api/kb/{args['kb_name']}/index-semantic") + + # Document Management + elif name == "upload_document": + return await api_post(f"/api/kb/{args['kb_name']}/import", { + "content": args["content"], + "filename": args["filename"], + "metadata": { + "title": args.get("title", ""), + "tags": args.get("tags", []) + } + }) + + elif name == "delete_document": + return await api_delete(f"/api/kb/{args['kb_name']}/documents/{args['filename']}") + + # Utility Tools + elif name == "get_ollama_models": + return await api_get("/api/ollama/models") + + elif name == "health_check": + try: + # Simple health check - list KBs should work if server is up + result = await api_get("/api/kb") + return {"status": "healthy", "message": "Claude OS is running", "kb_count": len(result.get("knowledge_bases", []))} + except Exception as e: + return {"status": "unhealthy", "error": str(e)} + + else: + return {"error": f"Unknown tool: {name}"} + + +async def main(): + """Run the MCP server.""" + async with stdio_server() as (read_stream, write_stream): + await server.run(read_stream, write_stream, server.create_initialization_options()) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py new file mode 100644 index 0000000..ffe95e5 --- /dev/null +++ b/tests/test_mcp_server.py @@ -0,0 +1,370 @@ +""" +Tests for the Claude Code MCP Server (claude_code_mcp.py). + +This is the critical bridge between Claude Code and the Claude OS REST API. +These tests ensure the MCP protocol is correctly implemented. +""" + +import pytest +import json +from unittest.mock import AsyncMock, patch, MagicMock +import sys +import os + +# Add the project root to the path +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + + +class TestMCPServerTools: + """Test the MCP server tool definitions.""" + + @pytest.fixture + def mock_server(self): + """Create a mock MCP server for testing.""" + with patch.dict(os.environ, {"CLAUDE_OS_API": "http://localhost:8051"}): + from mcp_server.claude_code_mcp import server, list_tools + return server, list_tools + + @pytest.mark.asyncio + async def test_list_tools_returns_all_tools(self, mock_server): + """Test that list_tools returns all expected tools.""" + server, list_tools_func = mock_server + tools = await list_tools_func() + + # Should have all the expected tools + tool_names = [t.name for t in tools] + + expected_tools = [ + "list_knowledge_bases", + "create_knowledge_base", + "search_knowledge_base", + "get_kb_stats", + "list_documents", + "delete_knowledge_base", + "list_projects", + "create_project", + "get_project", + "index_structural", + "index_semantic", + "upload_document", + "delete_document", + "get_ollama_models", + "health_check" + ] + + for expected in expected_tools: + assert expected in tool_names, f"Missing tool: {expected}" + + @pytest.mark.asyncio + async def test_list_tools_have_valid_schemas(self, mock_server): + """Test that all tools have valid input schemas.""" + server, list_tools_func = mock_server + tools = await list_tools_func() + + for tool in tools: + assert tool.inputSchema is not None, f"Tool {tool.name} has no schema" + assert "type" in tool.inputSchema, f"Tool {tool.name} schema missing 'type'" + assert tool.inputSchema["type"] == "object", f"Tool {tool.name} schema should be object" + assert "properties" in tool.inputSchema, f"Tool {tool.name} schema missing 'properties'" + assert "required" in tool.inputSchema, f"Tool {tool.name} schema missing 'required'" + + +class TestMCPServerAPIIntegration: + """Test the MCP server's integration with the REST API.""" + + @pytest.fixture + def mock_httpx(self): + """Mock httpx for API calls.""" + with patch("mcp_server.claude_code_mcp.httpx.AsyncClient") as mock: + yield mock + + @pytest.mark.asyncio + async def test_api_url_builder(self): + """Test the API URL builder function.""" + with patch.dict(os.environ, {"CLAUDE_OS_API": "http://test:9000"}): + # Re-import to pick up new env var + import importlib + import mcp_server.claude_code_mcp as mcp_module + importlib.reload(mcp_module) + + url = mcp_module.api_url("/api/kb") + assert url == "http://test:9000/api/kb" + + @pytest.mark.asyncio + async def test_list_knowledge_bases_calls_correct_endpoint(self, mock_httpx): + """Test that list_knowledge_bases calls the correct API endpoint.""" + mock_response = MagicMock() + mock_response.json.return_value = {"knowledge_bases": []} + mock_response.raise_for_status = MagicMock() + + mock_client = AsyncMock() + mock_client.get = AsyncMock(return_value=mock_response) + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=None) + mock_httpx.return_value = mock_client + + from mcp_server.claude_code_mcp import _execute_tool + result = await _execute_tool("list_knowledge_bases", {}) + + mock_client.get.assert_called_once() + call_args = mock_client.get.call_args + assert "/api/kb" in str(call_args) + + @pytest.mark.asyncio + async def test_search_knowledge_base_calls_correct_endpoint(self, mock_httpx): + """Test that search_knowledge_base calls the correct API endpoint.""" + mock_response = MagicMock() + mock_response.json.return_value = {"answer": "test", "sources": []} + mock_response.raise_for_status = MagicMock() + + mock_client = AsyncMock() + mock_client.post = AsyncMock(return_value=mock_response) + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=None) + mock_httpx.return_value = mock_client + + from mcp_server.claude_code_mcp import _execute_tool + result = await _execute_tool("search_knowledge_base", { + "kb_name": "test_kb", + "query": "test query" + }) + + mock_client.post.assert_called_once() + call_args = mock_client.post.call_args + assert "/api/kb/test_kb/chat" in str(call_args) + + @pytest.mark.asyncio + async def test_create_knowledge_base_sends_correct_payload(self, mock_httpx): + """Test that create_knowledge_base sends the correct payload.""" + mock_response = MagicMock() + mock_response.json.return_value = {"id": 1, "name": "new_kb"} + mock_response.raise_for_status = MagicMock() + + mock_client = AsyncMock() + mock_client.post = AsyncMock(return_value=mock_response) + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=None) + mock_httpx.return_value = mock_client + + from mcp_server.claude_code_mcp import _execute_tool + result = await _execute_tool("create_knowledge_base", { + "name": "new_kb", + "kb_type": "documentation", + "description": "Test KB" + }) + + mock_client.post.assert_called_once() + call_args = mock_client.post.call_args + # Check the JSON payload + assert call_args.kwargs.get("json", {}).get("name") == "new_kb" + + +class TestMCPServerErrorHandling: + """Test error handling in the MCP server.""" + + @pytest.mark.asyncio + async def test_unknown_tool_returns_error(self): + """Test that unknown tools return an error.""" + from mcp_server.claude_code_mcp import _execute_tool + result = await _execute_tool("nonexistent_tool", {}) + + assert "error" in result + assert "Unknown tool" in result["error"] + + @pytest.mark.asyncio + async def test_connection_error_handled_gracefully(self): + """Test that connection errors are handled gracefully.""" + import httpx + + with patch("mcp_server.claude_code_mcp.httpx.AsyncClient") as mock_httpx: + mock_client = AsyncMock() + mock_client.get = AsyncMock(side_effect=httpx.ConnectError("Connection refused")) + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=None) + mock_httpx.return_value = mock_client + + from mcp_server.claude_code_mcp import call_tool + result = await call_tool("list_knowledge_bases", {}) + + # Should return a TextContent with error message + assert len(result) == 1 + text = result[0].text + error_data = json.loads(text) + assert "error" in error_data + assert "Cannot connect" in error_data["error"] or "Connection" in str(error_data) + + @pytest.mark.asyncio + async def test_http_error_handled_gracefully(self): + """Test that HTTP errors are handled gracefully.""" + import httpx + + with patch("mcp_server.claude_code_mcp.httpx.AsyncClient") as mock_httpx: + mock_response = MagicMock() + mock_response.status_code = 500 + mock_response.text = "Internal Server Error" + + mock_client = AsyncMock() + mock_client.get = AsyncMock(side_effect=httpx.HTTPStatusError( + "Server error", + request=MagicMock(), + response=mock_response + )) + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=None) + mock_httpx.return_value = mock_client + + from mcp_server.claude_code_mcp import call_tool + result = await call_tool("list_knowledge_bases", {}) + + # Should return a TextContent with error message + assert len(result) == 1 + text = result[0].text + error_data = json.loads(text) + assert "error" in error_data + + +class TestMCPServerHealthCheck: + """Test the health check functionality.""" + + @pytest.mark.asyncio + async def test_health_check_success(self): + """Test health check when server is healthy.""" + with patch("mcp_server.claude_code_mcp.api_get") as mock_get: + mock_get.return_value = {"knowledge_bases": [{"id": 1}, {"id": 2}]} + + from mcp_server.claude_code_mcp import _execute_tool + result = await _execute_tool("health_check", {}) + + assert result["status"] == "healthy" + assert result["kb_count"] == 2 + + @pytest.mark.asyncio + async def test_health_check_failure(self): + """Test health check when server is unhealthy.""" + with patch("mcp_server.claude_code_mcp.api_get") as mock_get: + mock_get.side_effect = Exception("Connection failed") + + from mcp_server.claude_code_mcp import _execute_tool + result = await _execute_tool("health_check", {}) + + assert result["status"] == "unhealthy" + assert "error" in result + + +class TestMCPServerDocumentOperations: + """Test document-related operations.""" + + @pytest.fixture + def mock_httpx(self): + """Mock httpx for API calls.""" + with patch("mcp_server.claude_code_mcp.httpx.AsyncClient") as mock: + yield mock + + @pytest.mark.asyncio + async def test_upload_document_sends_correct_payload(self, mock_httpx): + """Test that upload_document sends the correct payload.""" + mock_response = MagicMock() + mock_response.json.return_value = {"success": True} + mock_response.raise_for_status = MagicMock() + + mock_client = AsyncMock() + mock_client.post = AsyncMock(return_value=mock_response) + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=None) + mock_httpx.return_value = mock_client + + from mcp_server.claude_code_mcp import _execute_tool + result = await _execute_tool("upload_document", { + "kb_name": "test_kb", + "content": "# Test Document\n\nThis is a test.", + "filename": "test.md", + "title": "Test Document", + "tags": ["test", "example"] + }) + + mock_client.post.assert_called_once() + call_args = mock_client.post.call_args + payload = call_args.kwargs.get("json", {}) + + assert payload.get("content") == "# Test Document\n\nThis is a test." + assert payload.get("filename") == "test.md" + assert payload.get("metadata", {}).get("title") == "Test Document" + + @pytest.mark.asyncio + async def test_delete_document_calls_correct_endpoint(self, mock_httpx): + """Test that delete_document calls the correct endpoint.""" + mock_response = MagicMock() + mock_response.json.return_value = {"success": True} + mock_response.raise_for_status = MagicMock() + + mock_client = AsyncMock() + mock_client.delete = AsyncMock(return_value=mock_response) + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=None) + mock_httpx.return_value = mock_client + + from mcp_server.claude_code_mcp import _execute_tool + result = await _execute_tool("delete_document", { + "kb_name": "test_kb", + "filename": "old_doc.md" + }) + + mock_client.delete.assert_called_once() + call_args = mock_client.delete.call_args + assert "/api/kb/test_kb/documents/old_doc.md" in str(call_args) + + +class TestMCPServerIndexingOperations: + """Test indexing-related operations.""" + + @pytest.fixture + def mock_httpx(self): + """Mock httpx for API calls.""" + with patch("mcp_server.claude_code_mcp.httpx.AsyncClient") as mock: + yield mock + + @pytest.mark.asyncio + async def test_index_structural_sends_path(self, mock_httpx): + """Test that index_structural sends the correct path.""" + mock_response = MagicMock() + mock_response.json.return_value = {"indexed": 50} + mock_response.raise_for_status = MagicMock() + + mock_client = AsyncMock() + mock_client.post = AsyncMock(return_value=mock_response) + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=None) + mock_httpx.return_value = mock_client + + from mcp_server.claude_code_mcp import _execute_tool + result = await _execute_tool("index_structural", { + "kb_name": "code_kb", + "path": "/path/to/project" + }) + + mock_client.post.assert_called_once() + call_args = mock_client.post.call_args + payload = call_args.kwargs.get("json", {}) + assert payload.get("path") == "/path/to/project" + + @pytest.mark.asyncio + async def test_index_semantic_calls_correct_endpoint(self, mock_httpx): + """Test that index_semantic calls the correct endpoint.""" + mock_response = MagicMock() + mock_response.json.return_value = {"embedded": 100} + mock_response.raise_for_status = MagicMock() + + mock_client = AsyncMock() + mock_client.post = AsyncMock(return_value=mock_response) + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=None) + mock_httpx.return_value = mock_client + + from mcp_server.claude_code_mcp import _execute_tool + result = await _execute_tool("index_semantic", { + "kb_name": "docs_kb" + }) + + mock_client.post.assert_called_once() + call_args = mock_client.post.call_args + assert "/api/kb/docs_kb/index-semantic" in str(call_args)