MCP Server Enhancement

MCP Server Enhancement Skill

This skill provides a systematic approach to extending the AI Counsel MCP server (server.py) with new tools while maintaining protocol compliance, stdio safety, and proper error handling.

Architecture Overview

The AI Counsel MCP server communicates via stdio (stdin/stdout) using the Model Context Protocol. Key architectural constraints:

• Stdio Safety: stdout is RESERVED for MCP protocol JSON. All logging MUST go to file (mcp_server.log) or stderr
• Protocol Compliance: Tools must follow MCP specification for request/response format
• Type Safety: Use Pydantic models for all request/response validation
• Error Isolation: Tool failures should return structured error responses, not crash the server
• Async First: All tool handlers are async functions using asyncio

Current Tool Architecture

Tool 1: deliberate (Primary Tool)

• Purpose: Multi-round AI model deliberation with consensus building
• Handler: call_tool() function (lines 242-327 in server.py)
• Request Model: DeliberateRequest (models/schema.py)
• Response Model: DeliberationResult (models/schema.py)
• Engine: Uses DeliberationEngine.execute() for orchestration

Tool 2: query_decisions (Decision Graph Tool)

• Purpose: Search and analyze past deliberations in decision graph memory
• Handler: handle_query_decisions() function (lines 329-415 in server.py)
• Request Schema: Inline in list_tools() (lines 196-237)
• Response: Custom JSON structure (not a Pydantic model)
• Conditional: Only exposed if config.decision_graph.enabled == True

Step-by-Step: Adding a New MCP Tool

Step 1: Define Pydantic Request/Response Models

Location: models/schema.py

Create type-safe models for your tool's inputs and outputs:

# In models/schema.py

class NewToolRequest(BaseModel):
    """Model for new_tool request."""

    parameter1: str = Field(
        ...,
        min_length=1,
        description="Description of parameter1"
    )
    parameter2: int = Field(
        default=5,
        ge=1,
        le=10,
        description="Integer parameter with range validation"
    )
    optional_param: Optional[str] = Field(
        default=None,
        description="Optional parameter"
    )

class NewToolResponse(BaseModel):
    """Model for new_tool response."""

    status: Literal["success", "partial", "failed"] = Field(
        ...,
        description="Operation status"
    )
    result_data: str = Field(..., description="Main result data")
    metadata: dict = Field(default_factory=dict, description="Additional metadata")

Best Practices:

• Use Field() with descriptive text for all fields (helps MCP client documentation)
• Use Literal types for enums (status fields, modes, etc.)
• Apply validation constraints (min_length, ge, le) at the model level
• Provide sensible defaults for optional parameters
• Use Optional[] for truly optional fields

Step 2: Add Tool Definition to list_tools()

Location: server.py, inside list_tools() function

Add your tool to the tools list returned by the MCP server:

@app.list_tools()
async def list_tools() -> list[Tool]:
    """List available MCP tools."""
    tools = [
        # Existing deliberate tool...
        Tool(
            name="deliberate",
            description=(...),
            inputSchema={...},
        ),

        # Your new tool
        Tool(
            name="new_tool",
            description=(
                "Clear, concise description of what this tool does. "
                "Include use cases and examples. Make it helpful for "
                "Claude Code users who will invoke this tool.\n\n"
                "Example usage:\n"
                '  {"parameter1": "example", "parameter2": 5}\n\n'
                "Expected behavior: Explain what the tool will do."
            ),
            inputSchema={
                "type": "object",
                "properties": {
                    "parameter1": {
                        "type": "string",
                        "description": "Description matching your Pydantic model",
                        "minLength": 1,
                    },
                    "parameter2": {
                        "type": "integer",
                        "description": "Integer parameter",
                        "minimum": 1,
                        "maximum": 10,
                        "default": 5,
                    },
                    "optional_param": {
                        "type": "string",
                        "description": "Optional parameter",
                    },
                },
                "required": ["parameter1"],  # Only required fields
            },
        ),
    ]

    return tools

Best Practices:

• inputSchema MUST match your Pydantic model (field names, types, constraints)
• Use JSON Schema types: string, integer, number, boolean, array, object
• Constraints: minLength, maxLength, minimum, maximum, minItems, maxItems
• Provide examples in the description (helps Claude Code understand usage)
• Multi-line descriptions are encouraged for clarity

Conditional Tools (like query_decisions):

# Add tool only if config enables it
if hasattr(config, "feature_name") and config.feature_name and config.feature_name.enabled:
    tools.append(
        Tool(name="conditional_tool", description=(...), inputSchema={...})
    )

Step 3: Create Tool Handler Function

Location: server.py, typically before main() function

Create an async handler function for your tool's logic:

async def handle_new_tool(arguments: dict) -> list[TextContent]:
    """
    Handle new_tool MCP tool call.

    Args:
        arguments: Tool arguments as dict (validated by MCP client)

    Returns:
        List of TextContent with JSON response

    Raises:
        Exception: Caught and converted to error response
    """
    try:
        # Step 1: Validate request with Pydantic
        logger.info(f"Validating new_tool request: {arguments}")
        request = NewToolRequest(**arguments)

        # Step 2: Execute your tool's logic
        logger.info(f"Processing new_tool: {request.parameter1}")

        # Example: Call engine or storage components
        # result_data = await some_engine.process(request.parameter1)
        result_data = f"Processed: {request.parameter1}"

        # Step 3: Build response model
        response_model = NewToolResponse(
            status="success",
            result_data=result_data,
            metadata={"parameter2_used": request.parameter2}
        )

        # Step 4: Serialize to JSON
        result_json = json.dumps(response_model.model_dump(), indent=2)
        logger.info(f"new_tool complete: {len(result_json)} chars")

        # Step 5: Return as TextContent
        return [TextContent(type="text", text=result_json)]

    except ValidationError as e:
        # Pydantic validation failure
        logger.error(f"Validation error in new_tool: {e}", exc_info=True)
        error_response = {
            "error": f"Invalid parameters: {str(e)}",
            "error_type": "ValidationError",
            "status": "failed",
        }
        return [TextContent(type="text", text=json.dumps(error_response, indent=2))]

    except Exception as e:
        # General error handling
        logger.error(f"Error in new_tool: {type(e).__name__}: {e}", exc_info=True)
        error_response = {
            "error": str(e),
            "error_type": type(e).__name__,
            "status": "failed",
        }
        return [TextContent(type="text", text=json.dumps(error_response, indent=2))]

Best Practices:

• Always use try-except to catch errors gracefully
• Log liberally to mcp_server.log (helps debugging)
• Return structured error responses (don't raise exceptions to MCP layer)
• Use Pydantic's model_dump() for serialization (ensures consistency)
• Separate validation errors from general errors for better diagnostics

Step 4: Route Tool Calls in call_tool()

Location: server.py, inside call_tool() function (around line 242)

Add routing logic to dispatch your new tool:

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    """
    Handle tool calls from MCP client.

    Args:
        name: Tool name
        arguments: Tool arguments as dict

    Returns:
        List of TextContent with JSON response
    """
    logger.info(f"Tool call received: {name} with arguments: {arguments}")

    # Route to appropriate handler
    if name == "new_tool":
        return await handle_new_tool(arguments)
    elif name == "query_decisions":
        return await handle_query_decisions(arguments)
    elif name == "deliberate":
        # Inline handler for deliberate (existing code)
        try:
            request = DeliberateRequest(**arguments)
            result = await engine.execute(request)
            # ... rest of deliberate logic
        except Exception as e:
            # ... error handling
    else:
        # Unknown tool error
        error_msg = f"Unknown tool: {name}"
        logger.error(error_msg)
        raise ValueError(error_msg)

Best Practices:

• Use early returns for clarity (avoid deep nesting)
• Keep routing logic simple (just dispatch, don't implement logic here)
• Log the tool name and arguments on entry (debugging aid)
• Raise ValueError for unknown tools (MCP client will handle gracefully)

Step 5: Write Tests

Location: Create tests/unit/test_new_tool.py and tests/integration/test_new_tool_integration.py

Unit Tests (Fast, No Dependencies)

# tests/unit/test_new_tool.py
import pytest
from models.schema import NewToolRequest, NewToolResponse
from pydantic import ValidationError

def test_new_tool_request_validation():
    """Test NewToolRequest validates correctly."""
    # Valid request
    req = NewToolRequest(parameter1="test", parameter2=7)
    assert req.parameter1 == "test"
    assert req.parameter2 == 7

    # Invalid: parameter2 out of range
    with pytest.raises(ValidationError):
        NewToolRequest(parameter1="test", parameter2=11)

    # Invalid: missing required parameter
    with pytest.raises(ValidationError):
        NewToolRequest(parameter2=5)

def test_new_tool_response_serialization():
    """Test NewToolResponse serializes correctly."""
    resp = NewToolResponse(
        status="success",
        result_data="test result",
        metadata={"key": "value"}
    )
    data = resp.model_dump()
    assert data["status"] == "success"
    assert data["result_data"] == "test result"
    assert data["metadata"]["key"] == "value"

Integration Tests (Real Server Invocation)

# tests/integration/test_new_tool_integration.py
import pytest
import json
from unittest.mock import AsyncMock, MagicMock
from mcp.types import TextContent

# Import your handler
from server import handle_new_tool

@pytest.mark.asyncio
async def test_handle_new_tool_success():
    """Test handle_new_tool with valid input."""
    arguments = {"parameter1": "test", "parameter2": 5}

    result = await handle_new_tool(arguments)

    assert len(result) == 1
    assert isinstance(result[0], TextContent)

    response_data = json.loads(result[0].text)
    assert response_data["status"] == "success"
    assert "test" in response_data["result_data"]

@pytest.mark.asyncio
async def test_handle_new_tool_validation_error():
    """Test handle_new_tool with invalid input."""
    arguments = {"parameter2": 5}  # Missing required parameter1

    result = await handle_new_tool(arguments)

    assert len(result) == 1
    response_data = json.loads(result[0].text)
    assert response_data["status"] == "failed"
    assert response_data["error_type"] == "ValidationError"

Testing Best Practices:

• Test validation (valid inputs, invalid inputs, edge cases)
• Test error handling (validation errors, runtime errors)
• Test serialization (model_dump() produces correct JSON)
• Mock external dependencies (engines, storage, API calls)
• Use pytest.mark.asyncio for async tests

Step 6: Update Documentation

Location: CLAUDE.md

Add your new tool to the architecture documentation:

## Architecture

### Core Components

**MCP Server Layer** (`server.py`)
- Entry point for MCP protocol communication via stdio
- Exposes tools: `deliberate`, `query_decisions`, `new_tool` (NEW)
- Tool: `new_tool` - [Brief description of what it does]

Update the data flow section if your tool has unique flow characteristics.

Critical Rules: Stdio Safety

WHY THIS MATTERS: The MCP server uses stdout for protocol communication. Any writes to stdout that aren't MCP protocol JSON will corrupt the communication channel and crash the server.

Rules

1. NEVER print() to stdout

   • Bad: print("Debug message")
   • Good: logger.info("Debug message")

2. NEVER write to sys.stdout

   • Bad: sys.stdout.write("output")
   • Good: sys.stderr.write("output") or use logger

3. Configure logging to file/stderr ONLY

   logging.basicConfig(
       handlers=[
           logging.FileHandler("mcp_server.log"),
           logging.StreamHandler(sys.stderr),  # NOT sys.stdout!
       ]
   )
   

4. Return MCP responses via TextContent

   • Good: return [TextContent(type="text", text=json.dumps(response))]
   • This is the ONLY correct way to send data to MCP client

5. Suppress subprocess stdout if not needed

   # If invoking external processes in your tool
   result = subprocess.run(
       ["command"],
       stdout=subprocess.PIPE,  # Capture, don't print
       stderr=subprocess.PIPE
   )
   

Testing Stdio Safety

Run your tool through the MCP client and verify:

• No garbled output in Claude Code
• Server log shows clean execution
• No "protocol error" messages from MCP client

Error Handling Patterns

Pattern 1: Pydantic Validation Errors

try:
    request = NewToolRequest(**arguments)
except ValidationError as e:
    logger.error(f"Validation error: {e}", exc_info=True)
    return [TextContent(type="text", text=json.dumps({
        "error": f"Invalid parameters: {str(e)}",
        "error_type": "ValidationError",
        "status": "failed",
    }, indent=2))]

Pattern 2: Runtime Errors

try:
    result = await some_operation()
except SomeSpecificError as e:
    logger.error(f"Operation failed: {e}", exc_info=True)
    return [TextContent(type="text", text=json.dumps({
        "error": str(e),
        "error_type": type(e).__name__,
        "status": "failed",
    }, indent=2))]

Pattern 3: Graceful Degradation

# If optional feature unavailable, return partial result
try:
    enhanced_data = await optional_enhancement()
except Exception as e:
    logger.warning(f"Enhancement failed, using base data: {e}")
    enhanced_data = None

return [TextContent(type="text", text=json.dumps({
    "status": "success" if enhanced_data else "partial",
    "result": base_data,
    "enhanced": enhanced_data,
}, indent=2))]

Pattern 4: Conditional Tool Availability

# In handle_new_tool()
if not hasattr(config, "feature") or not config.feature.enabled:
    return [TextContent(type="text", text=json.dumps({
        "error": "Feature not enabled in config.yaml",
        "error_type": "ConfigurationError",
        "status": "failed",
    }, indent=2))]

Integration with Existing Components

Using DeliberationEngine

If your tool needs to trigger deliberations:

from deliberation.engine import DeliberationEngine

async def handle_new_tool(arguments: dict) -> list[TextContent]:
    # Access global engine (initialized in server.py)
    request = DeliberateRequest(
        question="Generated question",
        participants=[...],
        rounds=2
    )
    result = await engine.execute(request)
    # Process result...

Using DecisionGraphStorage

If your tool needs to query decision graph:

from decision_graph.storage import DecisionGraphStorage
from pathlib import Path

async def handle_new_tool(arguments: dict) -> list[TextContent]:
    db_path = Path(config.decision_graph.db_path)
    if not db_path.is_absolute():
        db_path = PROJECT_DIR / db_path

    storage = DecisionGraphStorage(str(db_path))
    decisions = storage.get_all_decisions(limit=10)
    # Process decisions...

Using QueryEngine

If your tool needs advanced decision graph queries:

from deliberation.query_engine import QueryEngine

async def handle_new_tool(arguments: dict) -> list[TextContent]:
    engine = QueryEngine(storage)
    results = await engine.search_similar(query_text, limit=5)
    # Process results...

Configuration for New Tools

If your tool needs configuration, add to models/config.py and config.yaml:

In models/config.py:

class NewToolConfig(BaseModel):
    """Configuration for new_tool."""
    enabled: bool = Field(default=False, description="Enable new_tool feature")
    parameter: str = Field(default="default", description="Tool-specific parameter")
    timeout: int = Field(default=60, description="Timeout in seconds")

class Config(BaseModel):
    # ... existing config ...
    new_tool: Optional[NewToolConfig] = None

In config.yaml:

new_tool:
  enabled: true
  parameter: "custom_value"
  timeout: 120

Accessing config in handler:

async def handle_new_tool(arguments: dict) -> list[TextContent]:
    if not hasattr(config, "new_tool") or not config.new_tool.enabled:
        return error_response("new_tool not enabled")

    timeout = config.new_tool.timeout
    # Use config...

Testing Your New Tool End-to-End

1. Manual Testing via MCP Inspector

Use the MCP Inspector tool to test your tool directly:

# Install MCP Inspector
npm install -g @modelcontextprotocol/inspector

# Run inspector with your server
mcp-inspector python /path/to/server.py

Invoke your tool with test inputs and verify responses.

2. Integration with Claude Code

Add your server to ~/.claude/config/mcp.json:

{
  "mcpServers": {
    "ai-counsel": {
      "command": "python",
      "args": ["/path/to/ai-counsel/server.py"],
      "env": {}
    }
  }
}

Test in Claude Code:

1. Start a conversation
2. Claude Code should auto-discover your tool
3. Trigger your tool with a query that would use it
4. Verify the response is correct

3. Check Logs

Always check mcp_server.log after testing:

tail -f /path/to/ai-counsel/mcp_server.log

Look for:

• Tool invocation logs
• Validation successes/failures
• Error stack traces (if any)
• Performance timings

Common Pitfalls

Pitfall 1: inputSchema Mismatch with Pydantic Model

Problem: JSON Schema in list_tools() doesn't match Pydantic model fields.

Symptom: MCP client accepts invalid inputs, or rejects valid inputs.

Solution: Keep schemas in sync. Consider generating JSON Schema from Pydantic:

from pydantic.json_schema import JsonSchemaValue

schema = NewToolRequest.model_json_schema()
# Use this schema in inputSchema (but manually clean up for MCP if needed)

Pitfall 2: Forgetting to Route in call_tool()

Problem: Tool defined in list_tools() but not handled in call_tool().

Symptom: MCP client can invoke tool, but server returns "Unknown tool" error.

Solution: Always add routing in call_tool() after defining tool.

Pitfall 3: Blocking Operations in Handler

Problem: Tool handler does CPU-intensive or I/O-blocking work synchronously.

Symptom: Server becomes unresponsive, other tools timeout.

Solution: Use async operations or run blocking work in executor:

import asyncio

async def handle_new_tool(arguments: dict) -> list[TextContent]:
    # For CPU-bound work
    result = await asyncio.to_thread(blocking_function, arg1, arg2)

    # For I/O-bound work
    async with httpx.AsyncClient() as client:
        response = await client.get("https://api.example.com")

    # Process result...

Pitfall 4: Not Testing Error Cases

Problem: Only testing happy path, not validation failures or edge cases.

Symptom: Tool crashes or returns unclear errors when given bad input.

Solution: Write tests for every error scenario:

@pytest.mark.asyncio
async def test_handle_new_tool_errors():
    # Missing required field
    result = await handle_new_tool({})
    assert "ValidationError" in result[0].text

    # Invalid value range
    result = await handle_new_tool({"parameter1": "test", "parameter2": 999})
    assert "failed" in result[0].text

Checklist for Adding a New Tool

Use this checklist to ensure you've completed all steps:

• Define Pydantic request model in models/schema.py
• Define Pydantic response model in models/schema.py
• Add tool definition to list_tools() in server.py
• Ensure inputSchema matches Pydantic model exactly
• Create async handler function in server.py
• Add error handling (ValidationError + general exceptions)
• Add routing logic in call_tool()
• Write unit tests for models and validation
• Write integration tests for handler function
• Test stdio safety (no stdout contamination)
• Update CLAUDE.md architecture section
• Add configuration to models/config.py if needed
• Update config.yaml with default config if needed
• Test end-to-end with MCP Inspector
• Test integration with Claude Code
• Review logs for errors and performance
• Document any new dependencies in requirements.txt

References

• MCP Protocol Specification: https://spec.modelcontextprotocol.io/
• Pydantic Documentation: https://docs.pydantic.dev/
• MCP Python SDK: https://github.com/modelcontextprotocol/python-sdk
• AI Counsel Architecture: CLAUDE.md in repository root
• Existing Tool Implementations: server.py lines 104-416

Examples in Codebase

Study these existing implementations as reference:

1. Simple tool with inline handler: deliberate tool (lines 242-327 in server.py)

   • Shows: Pydantic validation, engine invocation, response truncation, error handling

2. Separate handler function: query_decisions tool (lines 329-415 in server.py)

   • Shows: Handler separation, conditional tool availability, storage integration

3. Conditional tool: Decision graph tools (lines 196-237 in server.py)

   • Shows: How to conditionally expose tools based on config

Getting Help

If you encounter issues:

1. Check mcp_server.log for detailed error traces
2. Verify stdio safety (no stdout writes)
3. Test Pydantic models in isolation first
4. Use MCP Inspector for manual testing before integration
5. Review existing tool implementations for patterns
6. Ensure all dependencies are installed (requirements.txt)

Remember: Stdio safety is paramount. When in doubt, log to file/stderr, NEVER stdout.