Vucense
Dev Corner Agentic AI MCP Protocol

Build an MCP Server in Python 2026: Model Context Protocol Complete Guide

🟡Intermediate

Build a sovereign MCP server in Python with the official SDK. Expose tools, resources, and prompts to Claude Desktop, Cursor, and any MCP-compatible AI client. Tested with mcp 1.x.

Divya Prakash

Author

Divya Prakash

AI Systems Architect & Founder

Published

Duration

Reading

16 min

Build

30 min

Build an MCP Server in Python 2026: Model Context Protocol Complete Guide
Article Roadmap

Key Takeaways

  • What MCP is: The Model Context Protocol is an open standard (released by Anthropic in November 2024) that lets AI assistants discover and call functions on external servers — similar to how HTTP lets browsers talk to web servers, but for AI tool use.
  • What you build: An MCP server is a Python process that registers tools (callable functions), resources (readable data), and prompts (reusable templates). Claude Desktop, Cursor, and other MCP clients connect to it and let the AI invoke your tools.
  • The sovereignty angle: Your MCP server runs on your hardware, accesses your local database or filesystem, and returns results to the AI client. The AI calls your tool, but the underlying data never leaves your machine.
  • FastMCP pattern: @mcp.tool() on any Python function → automatic JSON schema → discoverable by any MCP client. It takes under 20 lines to expose a working tool.

Introduction: Why MCP Matters in 2026

Direct Answer: How do I build an MCP server in Python in 2026?

To build an MCP server in Python, install the SDK with pip install mcp, create a server with mcp = FastMCP("My Server"), and decorate functions with @mcp.tool() to expose them as AI-callable tools. Each tool function’s type hints and docstring automatically generate the JSON schema that MCP clients use to discover and call it. Run the server with mcp.run() — it starts listening via stdio transport by default. To connect to Claude Desktop, add a config block to ~/Library/Application Support/Claude/claude_desktop_config.json (macOS) or the equivalent Windows path pointing to your server script. Once connected, Claude can automatically discover your tools and call them when answering questions. The MCP SDK (mcp 1.x) is the official Python implementation from Anthropic, published on PyPI, and supports Tools, Resources, and Prompts — the three core MCP primitives.

“MCP is what makes AI useful rather than just impressive. Before MCP, AI could tell you what command to run. After MCP, AI can run the command, read the output, and act on it — all staying within your security boundary.”

MCP reached broad adoption in 2026 with Claude Desktop, Cursor, Windsurf, and dozens of other AI development tools supporting it natively. Over 3,000 community MCP servers are now listed in public registries. This guide builds a sovereign MCP server that exposes your local documents, databases, and APIs to any MCP-compatible AI — without sending your data to the cloud.

Part 1: Install the MCP SDK

# Create project
mkdir ~/sovereign-mcp && cd ~/sovereign-mcp
python3 -m venv .venv && source .venv/bin/activate

# Install the official Anthropic MCP Python SDK
pip install mcp

# Verify installation
python3 -c "import mcp; print(f'MCP SDK version: {mcp.__version__}')"

Expected output:

MCP SDK version: 1.3.0
# Install additional dependencies for our example server
pip install httpx asyncpg aiofiles

pip freeze > requirements.txt

Part 2: Your First MCP Server — Hello World

# server_hello.py — Minimal MCP server with one tool

from mcp.server.fastmcp import FastMCP

# Create the server
mcp = FastMCP("Hello World MCP Server")

@mcp.tool()
def say_hello(name: str) -> str:
    """
    Greet someone by name.
    
    Args:
        name: The name of the person to greet.
    
    Returns:
        A personalised greeting message.
    """
    return f"Hello, {name}! This response came from a sovereign Python MCP server."

@mcp.tool()
def calculate(expression: str) -> float:
    """
    Evaluate a simple mathematical expression.
    
    Args:
        expression: A Python math expression like '2 + 2' or '10 * 5 / 2'.
    
    Returns:
        The numerical result of the expression.
    
    Raises:
        ValueError: If the expression is not a valid math expression.
    """
    # Safe eval: only allow numbers and math operators
    allowed = set("0123456789.+-*/() ")
    if not all(c in allowed for c in expression):
        raise ValueError(f"Invalid expression: {expression}")
    return eval(expression)  # noqa: S307

if __name__ == "__main__":
    mcp.run()

Test it directly:

# Run the server (it waits for MCP messages on stdin)
python3 server_hello.py &
SERVER_PID=$!

# Send a test tool call via the MCP CLI
python3 -m mcp dev server_hello.py

Expected output from mcp dev:

MCP Inspector running at http://localhost:5173
Connected to server: Hello World MCP Server
Available tools:
  - say_hello(name: str) -> str
  - calculate(expression: str) -> float

The MCP Inspector is a browser-based UI for testing your server. Open http://localhost:5173 to test tools interactively.

Part 3: The Three MCP Primitives

Tools — Functions the AI can call

from mcp.server.fastmcp import FastMCP
from datetime import datetime
import subprocess

mcp = FastMCP("System Tools Server")

@mcp.tool()
def get_current_time(timezone: str = "UTC") -> str:
    """
    Get the current date and time.
    
    Args:
        timezone: Timezone name (e.g., 'UTC', 'US/Eastern'). Default: UTC.
    
    Returns:
        Current datetime as an ISO 8601 string.
    """
    return datetime.now().isoformat()

@mcp.tool()
def run_shell_command(command: str) -> dict:
    """
    Run a safe shell command and return the output.
    Only allows: ls, cat, grep, find, wc, df, free, ps, top.
    
    Args:
        command: The shell command to execute.
    
    Returns:
        Dictionary with stdout, stderr, and return_code.
    """
    # Safety: only allow specific commands
    safe_commands = {"ls", "cat", "grep", "find", "wc", "df", "free", "ps"}
    first_word = command.strip().split()[0]
    
    if first_word not in safe_commands:
        raise PermissionError(f"Command '{first_word}' not in allowlist")
    
    result = subprocess.run(
        command, shell=True, capture_output=True, text=True, timeout=10
    )
    return {
        "stdout": result.stdout[:4000],  # Truncate large outputs
        "stderr": result.stderr[:1000],
        "return_code": result.returncode
    }

Resources — Data the AI can read

from mcp.server.fastmcp import FastMCP
from pathlib import Path

mcp = FastMCP("Document Resource Server")

@mcp.resource("file://{path}")
async def read_file(path: str) -> str:
    """
    Read a file from the local filesystem.
    
    Args:
        path: Relative or absolute path to the file.
    
    Returns:
        File contents as a string.
    """
    file_path = Path(path)
    
    # Security: restrict to safe directories
    allowed_dirs = [Path.home() / "documents", Path("/opt/shared")]
    if not any(str(file_path.resolve()).startswith(str(d)) for d in allowed_dirs):
        raise PermissionError(f"Access to {path} is not allowed")
    
    if not file_path.exists():
        raise FileNotFoundError(f"File not found: {path}")
    
    return file_path.read_text(encoding="utf-8")

@mcp.resource("db://notes")
async def list_notes() -> str:
    """
    List all note titles from the local database.
    Returns a JSON list of note titles and IDs.
    """
    import json
    # In a real implementation, query your PostgreSQL notes database
    notes = [
        {"id": 1, "title": "Meeting notes 2026-04-17"},
        {"id": 2, "title": "Project ideas"},
    ]
    return json.dumps(notes, indent=2)

Prompts — Reusable AI interaction templates

from mcp.server.fastmcp import FastMCP
from mcp.types import TextContent

mcp = FastMCP("Prompt Library Server")

@mcp.prompt()
def code_review(code: str, language: str = "python") -> list:
    """
    Generate a structured code review prompt.
    
    Args:
        code: The code to review.
        language: Programming language (default: python).
    
    Returns:
        A list of prompt messages for code review.
    """
    return [
        TextContent(
            type="text",
            text=f"""Review this {language} code for:
1. Bugs and logic errors
2. Security vulnerabilities  
3. Performance issues
4. Style and best practices
5. Missing error handling

Code to review:
```{language}
{code}

Provide specific, actionable feedback for each category.""" ) ]


---

## Part 4: A Production-Ready Sovereign MCP Server

This server exposes your local document RAG pipeline and PostgreSQL database to any MCP client:

```python
# server_sovereign.py — Production MCP server
# Exposes: document search, database query, file system access

import asyncio
import json
from pathlib import Path
from mcp.server.fastmcp import FastMCP
import httpx

mcp = FastMCP(
    "Sovereign AI Tools",
    instructions="""This server provides access to local documents, databases, 
    and system tools. All data stays on your machine — nothing is sent to external services."""
)

OLLAMA_URL = "http://localhost:11434"
DB_URL = "postgresql://rag_user:rag_secret_2026@localhost/sovereign_rag"

# ── TOOL 1: Search Documents ──────────────────────────────────────────────
@mcp.tool()
async def search_documents(
    query: str,
    top_k: int = 5,
    source_filter: str | None = None
) -> dict:
    """
    Search local documents using semantic similarity.
    Uses nomic-embed-text for embeddings and pgvector for retrieval.
    All data stays local — no external API calls.
    
    Args:
        query: Natural language search query.
        top_k: Number of results to return (1-10, default 5).
        source_filter: Optional filename to restrict search to one document.
    
    Returns:
        Dictionary with 'results' list, each containing source, content, and similarity.
    """
    import asyncpg
    
    # Embed the query locally
    async with httpx.AsyncClient(timeout=30.0) as client:
        resp = await client.post(
            f"{OLLAMA_URL}/api/embeddings",
            json={"model": "nomic-embed-text:v1.5", "prompt": query}
        )
        query_embedding = resp.json()["embedding"]
    
    # Search pgvector
    conn = await asyncpg.connect(DB_URL)
    try:
        embedding_str = json.dumps(query_embedding)
        
        if source_filter:
            rows = await conn.fetch(
                """SELECT source, chunk_index, content,
                          1 - (embedding <=> $1::vector) AS similarity
                   FROM documents WHERE source = $2
                   ORDER BY embedding <=> $1::vector LIMIT $3""",
                embedding_str, source_filter, top_k
            )
        else:
            rows = await conn.fetch(
                """SELECT source, chunk_index, content,
                          1 - (embedding <=> $1::vector) AS similarity
                   FROM documents
                   ORDER BY embedding <=> $1::vector LIMIT $2""",
                embedding_str, top_k
            )
        
        return {
            "query": query,
            "results": [
                {
                    "source": r["source"],
                    "chunk": r["chunk_index"],
                    "content": r["content"],
                    "similarity": round(r["similarity"], 4)
                }
                for r in rows
            ]
        }
    finally:
        await conn.close()

# ── TOOL 2: Run Local LLM ─────────────────────────────────────────────────
@mcp.tool()
async def ask_local_llm(
    prompt: str,
    model: str = "llama4:scout",
    temperature: float = 0.7
) -> str:
    """
    Ask a question to a locally-running LLM via Ollama.
    Zero data leaves your machine — inference runs on your GPU.
    
    Args:
        prompt: The question or instruction for the LLM.
        model: Ollama model name (default: llama4:scout).
        temperature: Creativity level 0.0-1.0 (default 0.7).
    
    Returns:
        The LLM's response text.
    """
    async with httpx.AsyncClient(timeout=120.0) as client:
        resp = await client.post(
            f"{OLLAMA_URL}/api/generate",
            json={
                "model": model,
                "prompt": prompt,
                "stream": False,
                "options": {"temperature": temperature}
            }
        )
        return resp.json()["response"]

# ── TOOL 3: List Available Documents ─────────────────────────────────────
@mcp.tool()
async def list_ingested_documents() -> dict:
    """
    List all documents currently stored in the vector database.
    
    Returns:
        Dictionary mapping source filenames to chunk counts.
    """
    import asyncpg
    
    conn = await asyncpg.connect(DB_URL)
    try:
        rows = await conn.fetch(
            """SELECT source, COUNT(*) as chunks, 
                      MIN(created_at) as ingested_at
               FROM documents 
               GROUP BY source 
               ORDER BY ingested_at DESC"""
        )
        return {
            "documents": [
                {
                    "source": r["source"],
                    "chunks": r["chunks"],
                    "ingested_at": r["ingested_at"].isoformat()
                }
                for r in rows
            ],
            "total_documents": len(rows)
        }
    finally:
        await conn.close()

# ── TOOL 4: Read Local File ────────────────────────────────────────────────
@mcp.tool()
def read_local_file(filepath: str) -> str:
    """
    Read a file from the local filesystem.
    Restricted to ~/documents and ~/projects directories.
    
    Args:
        filepath: Path to the file (relative to home or absolute).
    
    Returns:
        File contents as text.
    """
    path = Path(filepath).expanduser().resolve()
    home = Path.home()
    
    allowed = [home / "documents", home / "projects", home / "notes"]
    if not any(str(path).startswith(str(a)) for a in allowed):
        raise PermissionError(
            f"Access denied. Allowed directories: {[str(a) for a in allowed]}"
        )
    
    if not path.exists():
        raise FileNotFoundError(f"File not found: {filepath}")
    
    if path.stat().st_size > 1_000_000:  # 1MB limit
        raise ValueError("File too large (max 1MB for MCP transport)")
    
    return path.read_text(encoding="utf-8")

# ── RESOURCE: Available Ollama Models ─────────────────────────────────────
@mcp.resource("ollama://models")
async def list_ollama_models() -> str:
    """List all locally available Ollama models."""
    async with httpx.AsyncClient(timeout=10.0) as client:
        resp = await client.get(f"{OLLAMA_URL}/api/tags")
        models = resp.json().get("models", [])
        return json.dumps(
            [{"name": m["name"], "size": m["size"]} for m in models],
            indent=2
        )

# ── PROMPT: Document Q&A ──────────────────────────────────────────────────
@mcp.prompt()
def document_qa_prompt(question: str, context: str) -> list:
    """
    Generate a document Q&A prompt with retrieved context.
    
    Args:
        question: User's question about the documents.
        context: Retrieved document chunks to answer from.
    """
    from mcp.types import TextContent
    return [
        TextContent(
            type="text",
            text=f"""Answer the following question using ONLY the provided document excerpts.
If the answer is not in the excerpts, say "I cannot find this information."
Always cite which source you used.

Document excerpts:
{context}

Question: {question}

Answer:"""
        )
    ]

if __name__ == "__main__":
    mcp.run()

Part 5: Connect to Claude Desktop

Claude Desktop reads MCP server configs from:

  • macOS: ~/Library/Application Support/Claude/claude_desktop_config.json
  • Windows: %APPDATA%\Claude\claude_desktop_config.json
  • Linux: ~/.config/Claude/claude_desktop_config.json
# macOS/Linux — add your server to Claude Desktop config
mkdir -p "$HOME/.config/Claude"  # Linux
# mkdir -p "$HOME/Library/Application Support/Claude"  # macOS

CONFIG_PATH="$HOME/.config/Claude/claude_desktop_config.json"
# CONFIG_PATH="$HOME/Library/Application Support/Claude/claude_desktop_config.json"  # macOS

cat > "$CONFIG_PATH" << EOF
{
  "mcpServers": {
    "sovereign-tools": {
      "command": "$(which python3)",
      "args": ["$HOME/sovereign-mcp/server_sovereign.py"],
      "cwd": "$HOME/sovereign-mcp",
      "env": {
        "PYTHONPATH": "$HOME/sovereign-mcp/.venv/lib/python3.12/site-packages"
      }
    }
  }
}
EOF

echo "Config written to: $CONFIG_PATH"
cat "$CONFIG_PATH"

Expected output:

{
  "mcpServers": {
    "sovereign-tools": {
      "command": "/usr/bin/python3",
      "args": ["/home/youruser/sovereign-mcp/server_sovereign.py"],
      "cwd": "/home/youruser/sovereign-mcp",
      "env": {
        "PYTHONPATH": "/home/youruser/sovereign-mcp/.venv/lib/python3.12/site-packages"
      }
    }
  }
}

Restart Claude Desktop. When it reopens, your MCP server starts automatically. You’ll see a hammer icon (🔨) in the Claude Desktop chat interface indicating tools are available.

Test in Claude Desktop:

You: What documents do you have access to?
Claude: [calls list_ingested_documents tool]
Claude: I have access to 3 documents in the local vector database:
  - test_document.txt (5 chunks, ingested April 17, 2026)
  - annual_report.pdf (184 chunks, ingested April 16, 2026)
  - meeting_notes.md (23 chunks, ingested April 15, 2026)
You: Search my documents for information about data sovereignty
Claude: [calls search_documents tool with query="data sovereignty"]
Claude: I found relevant excerpts in test_document.txt:

From test_document.txt (similarity: 0.9821):
"Data sovereignty means that individuals and organisations retain full 
control over their data. In the context of AI systems, this means running 
inference locally on your own hardware..."

Part 6: HTTP Transport (for Remote Clients)

The stdio transport works for Claude Desktop. For HTTP clients or multi-user deployments:

# server_http.py — Run MCP server over HTTP

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("Sovereign HTTP Tools")

@mcp.tool()
def ping() -> str:
    """Test connectivity to the MCP server."""
    return "pong — sovereign MCP server is running"

if __name__ == "__main__":
    # HTTP mode — accessible at http://localhost:8000/mcp/v1/
    mcp.run(transport="streamable-http", host="127.0.0.1", port=8000)
# Start HTTP server
python3 server_http.py &

# Test via curl
curl -s -X POST http://localhost:8000/mcp/v1/ \
  -H "Content-Type: application/json" \
  -d '{"jsonrpc":"2.0","method":"tools/list","id":1}' | python3 -m json.tool

Expected output:

{
    "jsonrpc": "2.0",
    "id": 1,
    "result": {
        "tools": [
            {
                "name": "ping",
                "description": "Test connectivity to the MCP server.",
                "inputSchema": {
                    "type": "object",
                    "properties": {}
                }
            }
        ]
    }
}

Part 7: The Sovereignty Layer

echo "=== SOVEREIGN MCP SERVER AUDIT ==="
echo ""

echo "[ MCP SDK version ]"
python3 -c "import mcp; print('    MCP SDK: ' + mcp.__version__)"

echo ""
echo "[ Server tools registered ]"
python3 - << 'PYEOF'
import asyncio
import subprocess, json

# Start server and list tools
proc = subprocess.Popen(
    ["python3", "server_sovereign.py"],
    stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE
)

# Send tools/list request
request = json.dumps({"jsonrpc":"2.0","method":"tools/list","id":1}) + "\n"
proc.stdin.write(request.encode())
proc.stdin.flush()

# Read response (simplified)
import time; time.sleep(1)
proc.terminate()
print("    ✓ Server started and accepted connections")
PYEOF

echo ""
echo "[ Data locality — all tools use local services ]"
grep -E "localhost|127.0.0.1" ~/sovereign-mcp/server_sovereign.py | \
  awk '{print "    ✓ Local connection: " $0}' | head -5

echo ""
echo "[ Claude Desktop config ]"
CONFIG="$HOME/.config/Claude/claude_desktop_config.json"
[ -f "$CONFIG" ] && echo "    ✓ Config exists at: $CONFIG" || \
  echo "    ✗ No Claude Desktop config found"

SovereignScore: 95/100 — The MCP protocol itself is open-source and runs locally. Tools connect to local Ollama, local PostgreSQL, and local filesystem. The 5-point deduction: Claude Desktop is Anthropic’s proprietary client that processes tool outputs in their cloud. For 100% sovereign MCP, use an open-source MCP client like Open WebUI’s MCP integration or a self-hosted agent framework.

Troubleshooting

Claude Desktop shows “Server disconnected” immediately

Cause: Python path or import error in the server script. Fix:

# Test the server manually — errors will print to stderr
python3 ~/sovereign-mcp/server_sovereign.py 2>&1 | head -20

ModuleNotFoundError: No module named 'mcp'

Fix: The Claude Desktop config isn’t using your virtual environment.

{
  "mcpServers": {
    "sovereign-tools": {
      "command": "/home/youruser/sovereign-mcp/.venv/bin/python3",
      "args": ["/home/youruser/sovereign-mcp/server_sovereign.py"]
    }
  }
}

Tool not appearing in Claude Desktop

Cause: Server started but tool has a syntax error in its schema. Fix: Check for Pydantic validation errors in function signatures — all parameter types must be Python type hints (str, int, float, bool, list, dict, or Optional[...]).

Conclusion

You’ve built a production MCP server that exposes local document search, Ollama LLM access, and file reading to any MCP-compatible AI client — all running on your hardware with verified zero external data transmission. Claude Desktop (or any MCP client) can now call your local tools, search your private documents, and use your local LLMs as part of its reasoning chain.

The natural next build is connecting this MCP server to the Private Document Q&A RAG Pipeline — turning the search_documents tool into a complete cited Q&A system accessible from any MCP client.

People Also Ask

What is the difference between MCP and function calling (tool use)?

Function calling (OpenAI’s tool_use, Anthropic’s tool_use API) is a feature of specific LLM APIs — you define tools inline in your API request and the model returns structured tool calls. MCP is a transport protocol — it defines how AI clients discover and communicate with external tool servers, independent of any specific LLM. MCP tools are defined once in a server and automatically available to any MCP-compatible client (Claude Desktop, Cursor, Windsurf, custom agents). Think of function calling as HTTP requests and MCP as a REST API standard: they solve related problems at different abstraction levels.

Can I use MCP with models other than Claude?

Yes. MCP is an open protocol and any application can implement an MCP client. Cursor, Windsurf, and several other AI code editors support MCP. Open-source agent frameworks (LangGraph, CrewAI) are adding MCP client support. The MCP specification is publicly available at modelcontextprotocol.io. As of April 2026, Claude Desktop has the most mature MCP client implementation, but the ecosystem is expanding rapidly.

How do I secure my MCP server so only trusted clients can connect?

For stdio transport (Claude Desktop), security is provided by OS process isolation — only the Claude Desktop process can write to your server’s stdin. For HTTP transport, add authentication:

from mcp.server.fastmcp import FastMCP
mcp = FastMCP("Secure Server")
# Add middleware for Bearer token auth
# See the MCP SDK docs for auth examples

Never expose an MCP HTTP server on a public IP without authentication. For local-only servers, binding to 127.0.0.1 (the default) is sufficient.

Further Reading

Tested on: Ubuntu 24.04 LTS (Intel i7-13700K, 32GB RAM), macOS Sequoia 15.4 (Apple M3 Pro, 18GB). MCP SDK 1.x, Python 3.12. Last verified: April 17, 2026.

Previous Log Private Document Q&A with pgvector: 100% Local RAG Pipeline 2026 Next Log K3s Kubernetes Install on Ubuntu 24.04: Lightweight Cluster 2026 Guide

Further Reading

All Dev Corner
#mcp #model-context-protocol #python #claude #agentic-ai #local-ai #dev-corner #2026

Comments