Python MCP Done Right: Tools, Router, and LLM Integration with FastAPI¶

# MCP enables secure, controlled tool execution
# Tool servers expose capabilities with strict boundaries
# Router enforces policy and provides observability
# LLM gets safe, auditable tool access

# Example: Secure file reading
def read_file_safely(path: str) -> str:
    # MCP server validates path, enforces jail
    # Router checks permissions and rate limits
    # LLM gets controlled file access
    pass

# mcp_server.py - Minimal MCP server with filesystem and SQL tools
import sys
import json
import os
import sqlite3
import time
from typing import Any, Dict

# Configuration
BASE_DIR = os.environ.get("MCP_FS_BASE", "/app/readonly")  # jail
DB_PATH = os.environ.get("MCP_SQLITE", "/app/demo.sqlite")
HARD_TIMEOUT_MS = 4000

# Tool schema definitions
SCHEMA = {
    "tools": {
        "fs.read_text": {
            "params": {
                "type": "object",
                "required": ["path"],
                "properties": {
                    "path": {"type": "string"}
                }
            },
            "returns": {
                "type": "object",
                "properties": {
                    "text": {"type": "string"}
                }
            }
        },
        "sql.query": {
            "params": {
                "type": "object",
                "required": ["sql", "params"],
                "properties": {
                    "sql": {"type": "string"},
                    "params": {"type": "array"}
                }
            },
            "returns": {
                "type": "object",
                "properties": {
                    "rows": {"type": "array"}
                }
            }
        }
    }
}

def jail(path: str) -> str:
    """Jail path to prevent directory traversal"""
    p = os.path.realpath(os.path.join(BASE_DIR, path.lstrip("/")))
    if not p.startswith(os.path.realpath(BASE_DIR) + os.sep):
        raise PermissionError("path escapes jail")
    return p

def handle(method: str, params: Dict[str, Any]) -> Dict[str, Any]:
    """Handle MCP method calls with security checks"""
    t0 = time.time()

    if method == "schema":
        return SCHEMA

    if method == "fs.read_text":
        p = jail(params["path"])
        with open(p, "r", encoding="utf-8") as f:
            data = f.read(1024 * 1024)  # 1MB cap
        return {"text": data}

    if method == "sql.query":
        sql = params["sql"]
        if ";" in sql.replace(";;", ""):  # naive guard against multi-statement
            raise ValueError("multi-statement not allowed")

        with sqlite3.connect(DB_PATH, timeout=2.0) as conn:
            conn.row_factory = sqlite3.Row
            cur = conn.execute(sql, params.get("params", []))
            rows = [dict(r) for r in cur.fetchall()]
        return {"rows": rows}

    raise ValueError(f"unknown method {method}")

def rpc_loop():
    """Main RPC loop for stdio communication"""
    for line in sys.stdin:
        try:
            req = json.loads(line)

            # Check request age
            if (time.time() * 1000) - req.get("ts_ms", 0) > HARD_TIMEOUT_MS:
                raise TimeoutError("request too old")

            result = handle(req["method"], req.get("params", {}))
            out = {"id": req.get("id"), "ok": True, "result": result}

        except Exception as e:
            out = {
                "id": req.get("id"),
                "ok": False,
                "error": type(e).__name__,
                "message": str(e)
            }

        sys.stdout.write(json.dumps(out) + "\n")
        sys.stdout.flush()

if __name__ == "__main__":
    rpc_loop()

# router_app.py - FastAPI MCP Router with policy enforcement
import asyncio
import json
import os
import time
import uuid
from typing import Dict, Any, Optional
from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel, Field

class InvokeReq(BaseModel):
    tool: str
    params: Dict[str, Any] = Field(default_factory=dict)
    caller: Optional[str] = None

class InvokeRes(BaseModel):
    request_id: str
    tool: str
    duration_ms: int
    result: Optional[Dict[str, Any]] = None
    error: Optional[str] = None

class StdioServer:
    """Manages stdio-based MCP server processes"""

    def __init__(self, cmd: list[str], name: str):
        self.cmd = cmd
        self.name = name
        self.proc: Optional[asyncio.Process] = None
        self.lock = asyncio.Lock()

    async def start(self):
        """Start the MCP server process"""
        if self.proc and self.proc.returncode is None:
            return

        self.proc = await asyncio.create_subprocess_exec(
            *self.cmd,
            stdin=asyncio.subprocess.PIPE,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE
        )

    async def call(self, method: str, params: Dict[str, Any], timeout: float = 5.0) -> Dict[str, Any]:
        """Call a method on the MCP server"""
        await self.start()
        assert self.proc and self.proc.stdin and self.proc.stdout

        req = {
            "id": str(uuid.uuid4()),
            "method": method,
            "params": params,
            "ts_ms": int(time.time() * 1000)
        }

        line = json.dumps(req) + "\n"

        async with self.lock:
            self.proc.stdin.write(line.encode())
            await self.proc.stdin.drain()
            fut = asyncio.create_task(self.proc.stdout.readline())

        try:
            raw = await asyncio.wait_for(fut, timeout=timeout)
        except asyncio.TimeoutError:
            raise TimeoutError(f"{self.name} timeout")

        resp = json.loads(raw.decode())
        if not resp.get("ok"):
            raise RuntimeError(f"{self.name} error: {resp.get('error')} {resp.get('message')}")

        return resp["result"]

# FastAPI application
app = FastAPI(title="MCP Router", version="1.0.0")

# Register MCP servers
FS = StdioServer(["python", "mcp_server.py"], "fs+sqlite")
SERVERS = {"fs.read_text": FS, "sql.query": FS}

# Policy configuration
ALLOWLIST = {
    "fs.read_text": {"paths": ["/app/readonly/"]},
    "sql.query": {"readonly": True}
}

def authorize(req: InvokeReq):
    """Authorize tool invocation based on policy"""
    if req.tool not in ALLOWLIST:
        raise HTTPException(403, "tool not allowed")

    # Additional per-tool authorization logic
    if req.tool == "fs.read_text":
        path = req.params.get("path", "")
        if not path.startswith("/app/readonly/"):
            raise HTTPException(403, "path not in allowed directory")

    if req.tool == "sql.query":
        sql = req.params.get("sql", "").upper()
        if not sql.startswith("SELECT"):
            raise HTTPException(403, "only SELECT queries allowed")

@app.post("/invoke", response_model=InvokeRes)
async def invoke(req: InvokeReq):
    """Invoke a tool through the MCP router"""
    authorize(req)

    server = SERVERS.get(req.tool)
    if not server:
        raise HTTPException(404, "tool not registered")

    t0 = time.time()
    request_id = str(uuid.uuid4())

    try:
        res = await server.call(req.tool, req.params, timeout=5.0)
        return InvokeRes(
            request_id=request_id,
            tool=req.tool,
            duration_ms=int((time.time() - t0) * 1000),
            result=res
        )
    except Exception as e:
        return InvokeRes(
            request_id=request_id,
            tool=req.tool,
            duration_ms=int((time.time() - t0) * 1000),
            error=str(e)
        )

@app.get("/tools")
async def list_tools():
    """List available tools and their schemas"""
    tools = {}
    for tool_name, server in SERVERS.items():
        try:
            schema = await server.call("schema", {})
            tools[tool_name] = schema["tools"][tool_name]
        except Exception as e:
            tools[tool_name] = {"error": str(e)}

    return {"tools": tools}

@app.get("/health")
async def health_check():
    """Health check endpoint"""
    return {"status": "healthy", "servers": len(SERVERS)}

# llm_orchestrator.py - LLM integration with function calling
import httpx
import json
import os
from typing import List, Dict, Any

# Tool definitions for LLM
TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "fs.read_text",
            "description": "Read a small UTF-8 file from the allowed project directory.",
            "parameters": {
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "File path relative to base directory"}
                },
                "required": ["path"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "sql.query",
            "description": "Run a SELECT query (read-only) against the demo SQLite database.",
            "parameters": {
                "type": "object",
                "properties": {
                    "sql": {"type": "string", "description": "SQL SELECT query"},
                    "params": {"type": "array", "description": "Query parameters", "items": {"type": "string"}}
                },
                "required": ["sql", "params"]
            }
        }
    }
]

class LLMOrchestrator:
    """Orchestrates LLM calls with MCP tool integration"""

    def __init__(self, router_url: str = "http://localhost:8000"):
        self.router_url = router_url
        self.client = httpx.AsyncClient(timeout=30.0)

    async def chat(self, messages: List[Dict[str, str]]) -> str:
        """Chat with LLM using MCP tools"""

        # 1) Call LLM with available tools
        async with httpx.AsyncClient(timeout=30) as client:
            response = await client.post(
                "https://api.openai.com/v1/chat/completions",
                headers={"Authorization": f"Bearer {os.environ['OPENAI_API_KEY']}"},
                json={
                    "model": "gpt-4o-mini",
                    "messages": messages,
                    "tools": TOOLS,
                    "tool_choice": "auto"
                }
            )
            data = response.json()

        # 2) Execute tool calls via MCP Router
        tool_calls = data["choices"][0]["message"].get("tool_calls", [])

        for call in tool_calls:
            name = call["function"]["name"]
            args = json.loads(call["function"]["arguments"])

            # Call MCP Router
            tool_response = await self.client.post(
                f"{self.router_url}/invoke",
                json={"tool": name, "params": args}
            )

            # Add tool result to messages
            tool_msg = {
                "role": "tool",
                "tool_call_id": call["id"],
                "name": name,
                "content": tool_response.text
            }
            messages.append(tool_msg)

        # 3) Final LLM call with tool results
        async with httpx.AsyncClient(timeout=30) as client:
            response = await client.post(
                "https://api.openai.com/v1/chat/completions",
                headers={"Authorization": f"Bearer {os.environ['OPENAI_API_KEY']}"},
                json={
                    "model": "gpt-4o-mini",
                    "messages": messages
                }
            )
            return response.json()["choices"][0]["message"]["content"]

# Example usage
async def main():
    orchestrator = LLMOrchestrator()

    messages = [
        {"role": "user", "content": "Read the file hello.txt and summarize its contents."}
    ]

    response = await orchestrator.chat(messages)
    print(f"LLM Response: {response}")

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

# ollama_integration.py - Local LLM integration with MCP
import httpx
import json
from typing import List, Dict, Any

class OllamaMCPOrchestrator:
    """Ollama integration with MCP tools (no function calling)"""

    def __init__(self, router_url: str = "http://localhost:8000", ollama_url: str = "http://localhost:11434"):
        self.router_url = router_url
        self.ollama_url = ollama_url
        self.client = httpx.AsyncClient()

    async def chat_with_tools(self, user_message: str, model: str = "llama3") -> str:
        """Chat with Ollama using MCP tools"""

        # 1) Analyze user message for tool needs
        tool_requests = self._analyze_tool_needs(user_message)

        # 2) Execute tools and collect results
        tool_results = []
        for tool_name, tool_args in tool_requests:
            try:
                response = await self.client.post(
                    f"{self.router_url}/invoke",
                    json={"tool": tool_name, "params": tool_args}
                )
                result = response.json()
                tool_results.append(f"{tool_name}: {result}")
            except Exception as e:
                tool_results.append(f"{tool_name}: Error - {e}")

        # 3) Build context with tool results
        context = f"""
        User request: {user_message}

        Tool results:
        {chr(10).join(tool_results)}

        Please respond to the user's request using the tool results above.
        """

        # 4) Call Ollama with context
        response = await self.client.post(
            f"{self.ollama_url}/api/generate",
            json={
                "model": model,
                "prompt": context,
                "stream": False
            }
        )

        return response.json()["response"]

    def _analyze_tool_needs(self, message: str) -> List[tuple]:
        """Simple heuristic to determine tool needs"""
        tool_requests = []

        # Check for file reading requests
        if any(word in message.lower() for word in ["read", "file", "content", "text"]):
            # Extract potential file path
            words = message.split()
            for i, word in enumerate(words):
                if word.endswith(".txt") or word.endswith(".md"):
                    tool_requests.append(("fs.read_text", {"path": word}))

        # Check for database queries
        if any(word in message.lower() for word in ["query", "database", "sql", "table", "data"]):
            tool_requests.append(("sql.query", {"sql": "SELECT * FROM sqlite_master", "params": []}))

        return tool_requests

# Example usage
async def main():
    orchestrator = OllamaMCPOrchestrator()

    response = await orchestrator.chat_with_tools("Read hello.txt and tell me what it contains")
    print(f"Ollama Response: {response}")

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

# production_config.py - Production configuration and monitoring
import os
import logging
from typing import Dict, Any
from prometheus_client import Counter, Histogram, Gauge

# Metrics
tool_invocations = Counter('mcp_tool_invocations_total', 'Total tool invocations', ['tool', 'status'])
tool_duration = Histogram('mcp_tool_duration_seconds', 'Tool execution duration', ['tool'])
active_connections = Gauge('mcp_active_connections', 'Active MCP connections')

class ProductionConfig:
    """Production configuration for MCP systems"""

    def __init__(self):
        self.max_file_size = int(os.getenv("MCP_MAX_FILE_SIZE", "1048576"))  # 1MB
        self.max_sql_rows = int(os.getenv("MCP_MAX_SQL_ROWS", "1000"))
        self.request_timeout = int(os.getenv("MCP_REQUEST_TIMEOUT", "30"))
        self.max_concurrent = int(os.getenv("MCP_MAX_CONCURRENT", "10"))

        # Security settings
        self.allowed_domains = os.getenv("MCP_ALLOWED_DOMAINS", "").split(",")
        self.jwt_secret = os.getenv("MCP_JWT_SECRET", "change-me-in-production")
        self.audit_log_path = os.getenv("MCP_AUDIT_LOG_PATH", "/var/log/mcp-audit.log")

        # Monitoring
        self.metrics_enabled = os.getenv("MCP_METRICS_ENABLED", "true").lower() == "true"
        self.log_level = os.getenv("MCP_LOG_LEVEL", "INFO")

        self._setup_logging()

    def _setup_logging(self):
        """Setup structured logging"""
        logging.basicConfig(
            level=getattr(logging, self.log_level),
            format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
            handlers=[
                logging.StreamHandler(),
                logging.FileHandler(self.audit_log_path)
            ]
        )

    def validate_request(self, tool: str, params: Dict[str, Any]) -> bool:
        """Validate request against production limits"""

        # File size limits
        if tool == "fs.read_text":
            path = params.get("path", "")
            if len(path) > 255:  # Path length limit
                return False

        # SQL query limits
        if tool == "sql.query":
            sql = params.get("sql", "")
            if len(sql) > 10000:  # SQL length limit
                return False
            if not sql.strip().upper().startswith("SELECT"):
                return False

        return True

    def log_audit(self, request_id: str, tool: str, params: Dict[str, Any], result: Any):
        """Log audit trail"""
        audit_entry = {
            "request_id": request_id,
            "tool": tool,
            "params": params,
            "result_size": len(str(result)) if result else 0,
            "timestamp": time.time()
        }

        logging.info(f"AUDIT: {json.dumps(audit_entry)}")

# ❌ WRONG: Insecure tool implementation
def bad_file_reader(path: str) -> str:
    # No path validation - directory traversal vulnerability
    with open(path, 'r') as f:
        return f.read()  # No size limit - memory exhaustion

# ✅ CORRECT: Secure tool implementation
def good_file_reader(path: str) -> str:
    # Path jailing prevents traversal
    safe_path = jail_path(path)

    # Size limit prevents memory exhaustion
    with open(safe_path, 'r') as f:
        return f.read(1024 * 1024)  # 1MB limit

# ❌ WRONG: Inefficient tool routing
def bad_tool_router(tool: str, params: dict):
    # No connection pooling
    # No timeout handling
    # No error recovery
    response = requests.post(f"http://tool-server/{tool}", json=params)
    return response.json()

# ✅ CORRECT: Efficient tool routing
async def good_tool_router(tool: str, params: dict):
    # Connection pooling
    # Timeout handling
    # Error recovery with circuit breakers
    async with httpx.AsyncClient(timeout=5.0) as client:
        response = await client.post(f"http://tool-server/{tool}", json=params)
        return response.json()

# ✅ DO: Complete MCP implementation checklist
def mcp_implementation_checklist():
    checklist = {
        "tool_design": [
            "One responsibility per tool",
            "Explicit schemas and validation",
            "Idempotent operations where possible",
            "Time and size limits"
        ],
        "security": [
            "Path jailing for filesystem tools",
            "Domain allow-lists for HTTP tools",
            "Read-only roles for database tools",
            "Per-caller RBAC with JWT"
        ],
        "resilience": [
            "Circuit breakers around servers",
            "Retries with exponential backoff",
            "Kill hung servers automatically",
            "Hot reload schemas on restart"
        ],
        "observability": [
            "Request ID tracking",
            "Duration metrics per tool",
            "Error categorization",
            "Structured JSON logs"
        ]
    }
    return checklist

# ✅ DO: Production deployment checklist
def production_deployment_checklist():
    checklist = {
        "infrastructure": [
            "Container-based tool servers",
            "Read-only root filesystems",
            "Least-privilege mounts",
            "Network isolation"
        ],
        "monitoring": [
            "Prometheus metrics collection",
            "Structured logging with JSON",
            "Health check endpoints",
            "Alert on tool failures"
        ],
        "security": [
            "JWT authentication",
            "Per-tool ACLs",
            "Audit logging",
            "Resource limits"
        ],
        "performance": [
            "Connection pooling",
            "Request timeouts",
            "Concurrency limits",
            "Circuit breakers"
        ]
    }
    return checklist

# Essential MCP patterns
def essential_mcp_patterns():
    # 1. Tool server with stdio transport
    # 2. Router with policy enforcement
    # 3. LLM integration with function calling
    # 4. Production monitoring and security
    pass