22. Custom MCP Tool Suite Project

Chapter 22 of 22 · 25 min

KEY INSIGHT

Building a complete MCP tool suite integrates everything learnedΓÇöprotocol handling, security, testing, and observabilityΓÇöinto a cohesive, production-ready system. This capstone project creates an MCP-powered development assistant with file management, code search, and task execution capabilities. **Project Structure:** ``` dev-assistant-mcp/ Γö£ΓöÇΓöÇ pyproject.toml Γö£ΓöÇΓöÇ src/ Γöé Γö£ΓöÇΓöÇ __init__.py Γöé Γö£ΓöÇΓöÇ server.py # Main MCP server Γöé Γö£ΓöÇΓöÇ tools/ Γöé Γöé Γö£ΓöÇΓöÇ __init__.py Γöé Γöé Γö£ΓöÇΓöÇ files.py # File operations Γöé Γöé Γö£ΓöÇΓöÇ search.py # Code search Γöé Γöé ΓööΓöÇΓöÇ tasks.py # Task execution Γöé Γö£ΓöÇΓöÇ auth/ Γöé Γöé Γö£ΓöÇΓöÇ __init__.py Γöé Γöé ΓööΓöÇΓöÇ token.py # JWT authentication Γöé Γö£ΓöÇΓöÇ middleware/ Γöé Γöé Γö£ΓöÇΓöÇ __init__.py Γöé Γöé ΓööΓöÇΓöÇ logging.py # Request logging Γöé ΓööΓöÇΓöÇ tests/ Γöé Γö£ΓöÇΓöÇ __init__.py Γöé Γö£ΓöÇΓöÇ test_tools.py Γöé ΓööΓöÇΓöÇ test_auth.py ΓööΓöÇΓöÇ config/ ΓööΓöÇΓöÇ settings.yaml ``` **Core Implementation:** ```python # server.py from fastapi import FastAPI from mcp.server.fastmcp import FastMCP from contextlib import asynccontextmanager app = FastAPI() mcp = FastMCP("Dev Assistant") # Register tools from .tools.files import register_file_tools from .tools.search import register_search_tools from .tools.tasks import register_task_tools register_file_tools(mcp) register_search_tools(mcp) register_task_tools(mcp) # Add auth middleware from .auth.token import AuthMiddleware app.add_middleware(AuthMiddleware) @asynccontextmanager async def lifespan(app: FastAPI): # Startup logger.info("Starting Dev Assistant MCP server") yield # Shutdown logger.info("Shutting down Dev Assistant MCP server") app.router.lifespan_context = lifespan # Mount MCP at /mcp app.mount("/mcp", mcp.streamable_http_app()) ``` **File Tools:** ```python # tools/files.py from pathlib import Path from typing import Annotated from fastapi import Path as PathParam def register_file_tools(mcp: FastMCP): @mcp.tool(description="Read file contents safely") def read_file( path: Annotated[str, PathParam(description="Absolute file path")] ) -> str: p = Path(path).resolve() if not p.exists(): raise ToolError(f"File not found: {path}", code="not_found") if p.stat().st_size > 1_000_000: raise ToolError("File too large (max 1MB)", code="size_exceeded") return p.read_text() @mcp.tool(description="List directory contents") def list_directory( path: Annotated[str, PathParam(description="Directory path")], pattern: str = "*" ) -> list[dict]: p = Path(path).resolve() return [ {"name": f.name, "type": "dir" if f.is_dir() else "file"} for f in p.glob(pattern) ] @mcp.tool(description="Write content to file") def write_file( path: Annotated[str, PathParam(description="File path")], content: str ) -> dict: p = Path(path) p.parent.mkdir(parents=True, exist_ok=True) p.write_text(content) return {"written": len(content), "path": str(p)} ``` **Search Tools:** ```python # tools/search.py import subprocess import shlex def register_search_tools(mcp: FastMCP): @mcp.tool(description="Search files using grep") def grep( pattern: str, path: str = ".", file_pattern: str = "*" ) -> list[str]: safe_pattern = shlex.quote(pattern) safe_path = shlex.quote(path) result = subprocess.run( ["grep", "-r", "--include=" + file_pattern, safe_pattern, safe_path], capture_output=True, text=True, timeout=30 ) if result.returncode not in (0, 1): raise ToolError(result.stderr, code="search_failed") return result.stdout.strip().split("\n") if result.stdout else [] @mcp.tool(description="Search code structure") def find_symbol( name: str, kind: str = "function" # function, class, variable ) -> list[dict]: # Implementation using tree-sitter or similar pass ``` **Task Tools:** ```python # tools/tasks.py import subprocess def register_task_tools(mcp: FastMCP): @mcp.tool(description="Execute shell command with timeout") def execute( command: str, cwd: str = None, timeout: int = 60 ) -> dict: try: result = subprocess.run( command, shell=True, cwd=cwd, capture_output=True, text=True, timeout=timeout ) return { "exit_code": result.returncode, "stdout": result.stdout, "stderr": result.stderr, } except subprocess.TimeoutExpired: raise ToolError(f"Command timed out after {timeout}s", code="timeout") ``` **Authentication:** ```python # auth/token.py import jwt from starlette.middleware.base import BaseHTTPMiddleware class AuthMiddleware(BaseHTTPMiddleware): async def dispatch(self, request, call_next): # Skip auth for health endpoint if request.url.path == "/health": return await call_next(request) # Extract token auth = request.headers.get("Authorization", "") if not auth.startswith("Bearer "): return JSONResponse({"error": "Unauthorized"}, 401) token = auth[7:] try: payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"]) request.state.user = payload except jwt.ExpiredSignatureError: return JSONResponse({"error": "Token expired"}, 401) except jwt.InvalidTokenError: return JSONResponse({"error": "Invalid token"}, 401) return await call_next(request) ``` **Testing:** ```python # tests/test_tools.py import pytest @pytest.mark.asyncio async def test_read_file_success(): create_temp_file("/tmp/test.txt", "hello") result = await call_tool("read_file", {"path": "/tmp/test.txt"}) assert result == "hello" @pytest.mark.asyncio async def test_read_file_not_found(): with pytest.raises(ToolError) as exc: await call_tool("read_file", {"path": "/nonexistent"}) assert exc.value.code == "not_found" ``` **Running the Server:** ```bash # Development uv run dev # Production uv run serve --port 8000 --host 0.0.0.0 ```

Knowledge transfer checkpoint

Connect Custom MCP Tool Suite Project back to the local-AI decision you are learning to make. The practical question is not only whether the code or concept works, but whether it still works when the model, runtime, hardware budget, privacy requirement, and latency target are real constraints.

Before moving on, write down four things: the local runtime or deployment surface involved, the memory or throughput constraint that could change the design, the verification signal that proves the lesson worked, and the failure mode you would check first if the result looked wrong. That turns this chapter from background knowledge into an operator habit.

A good answer should be specific enough that another reader could repeat the decision on their own machine. Name the model or component when there is one, record the relevant context or token budget, and prefer a measurable check over a vague statement such as "it seems faster" or "the setup is fine."

EXERCISE

Extend this project with additional tools (git operations, database queries, documentation generation). Add thorough tests covering happy paths and error cases. Deploy with proper TLS and monitoring.