TL;DR

  • Three production MCP servers I run: a docs/wiki wrapper, a personal-finance aggregator, and an inventory-ops tool for a side project.
  • Type hints are your schema. The FastMCP SDK extracts JSON Schema from Python docstrings and type annotations — you almost never hand-write schemas.
  • Flatten nested APIs. If the upstream API returns nested JSON, unwrap it in the MCP tool and return clean markdown or a simple dict — agents prefer predictable structures.
  • Reads always-on, writes gated. Use environment variables (MCP_WRITE=1) or similar to gate mutation tools; put mutations behind a validated service layer and audit every call.
  • Log to stderr, not stdout. Stdio protocol uses stdout for JSON-RPC; anything else breaks the connection.

The lineup

I’ve written three MCP servers from scratch over the last few months. This is the cookbook — concrete patterns, not theory. If you haven’t read the intro to writing MCP servers, start there; I won’t re-explain the anatomy.

Each server wraps a different API shape and taught me something different:

  1. Wiki MCP — wraps a self-hosted wiki’s GraphQL API. Lesson: flatten nested results.
  2. Finance MCP — wraps Monarch Money’s REST API. Lesson: modularize tools by domain; keep auth outside.
  3. Inventory MCP — talks to a Postgres database behind a port-forward. Lesson: gate writes, audit everything, handle long-running operations.

Let’s walk each one.

Server 1: Wiki MCP

The problem: I have a self-hosted wiki (GraphQL API) with hundreds of pages. I want Claude to search for docs, read them, and sometimes create new pages or update existing ones.

The server: ~200 lines of Python, 6 tools.

import mcp.server
from mcp.types import Tool
from fastmcp import FastMCP
import requests

app = FastMCP("wiki-mcp")

@app.tool()
def search_wiki(query: str, limit: int = 10) -> str:
    """Search wiki pages by title or tag. Returns markdown list with URLs."""
    # Call GraphQL API, flatten results
    results = call_graphql_search(query)
    
    lines = [f"- **{page['title']}** (id={page['id']}) - {page['summary']}" 
             for page in results[:limit]]
    return "\n".join(lines)

@app.tool()
def read_page(page_id: str) -> str:
    """Read a wiki page by ID. Returns markdown content + metadata."""
    page = call_graphql_read(page_id)
    return f"""# {page['title']}

**Tags:** {', '.join(page['tags'])}
**Modified:** {page['updated_at']}

{page['content']}
"""

@app.tool()
def list_pages(tag: str | None = None) -> str:
    """List all pages, optionally filtered by tag."""
    pages = call_graphql_list(tag)
    return "\n".join(f"- {p['title']} ({p['id']})" for p in pages)

@app.tool()
def create_page(title: str, content: str, parent_id: str | None = None) -> str:
    """Create a new wiki page. Requires MCP_WRITE=1."""
    if not os.getenv("MCP_WRITE"):
        return "ERROR: Write mode disabled. Set MCP_WRITE=1 to enable."
    
    result = call_graphql_create(title, content, parent_id)
    return f"Created page {result['id']}: {result['url']}"

if __name__ == "__main__":
    mcp.run(transport="stdio", app=app)

Key insight: The underlying GraphQL API returns nested objects with 15+ fields per page. I flatten it — a search_wiki tool returns a markdown list, not JSON. When the agent reads a page, it gets markdown with a small header, not raw nested JSON. Flat structures = fewer hallucinations.

Server 2: Finance MCP

The problem: Monarch Money has a REST API with 40+ endpoints (accounts, transactions, budgets, rules, tags, splits, net worth). I want Claude to answer questions about my finances without needing to know the API shape.

The server: ~800 lines, organized into domain modules.

from fastmcp import FastMCP
import os
import json

app = FastMCP("finance-mcp")

# Auth is handled separately — a one-time setup script
# that does MFA and saves a session token to ~/.finance_session
def load_session():
    with open(os.path.expanduser("~/.finance_session")) as f:
        return json.load(f)

SESSION = load_session()

# Read tools — always available
@app.tool()
def list_accounts() -> str:
    """List all connected accounts (checking, savings, investment, etc)."""
    resp = requests.get(
        "https://api.monarchmoney.com/accounts",
        headers={"Authorization": f"Bearer {SESSION['token']}"}
    )
    accounts = resp.json()["data"]
    return "\n".join(
        f"- {a['name']} ({a['type']}): ${a['balance']:.2f}"
        for a in accounts
    )

@app.tool()
def get_net_worth() -> str:
    """Current net worth across all accounts."""
    resp = requests.get(
        "https://api.monarchmoney.com/net-worth",
        headers={"Authorization": f"Bearer {SESSION['token']}"}
    )
    nw = resp.json()["data"]
    return f"Net worth: ${nw['total']:.2f} (assets: ${nw['assets']:.2f}, liabilities: ${nw['liabilities']:.2f})"

@app.tool()
def list_transactions(account_id: str, limit: int = 20) -> str:
    """List recent transactions from an account."""
    resp = requests.get(
        f"https://api.monarchmoney.com/accounts/{account_id}/transactions",
        params={"limit": limit},
        headers={"Authorization": f"Bearer {SESSION['token']}"}
    )
    txns = resp.json()["data"]
    return "\n".join(
        f"- {t['date']}: {t['payee']} ${t['amount']:.2f} ({t['category']})"
        for t in txns
    )

# Write tool — gated and previewable
@app.tool()
def create_transaction(
    account_id: str,
    payee: str,
    amount: float,
    category: str,
    date: str,
    dry_run: bool = False
) -> str:
    """Create a transaction. Set dry_run=True to preview."""
    if not os.getenv("MCP_WRITE"):
        return "ERROR: Write mode disabled."
    
    payload = {
        "payee": payee,
        "amount": amount,
        "category": category,
        "date": date
    }
    
    if dry_run:
        return f"PREVIEW: Would create transaction:\n{json.dumps(payload, indent=2)}"
    
    resp = requests.post(
        f"https://api.monarchmoney.com/accounts/{account_id}/transactions",
        json=payload,
        headers={"Authorization": f"Bearer {SESSION['token']}"}
    )
    
    if resp.status_code != 201:
        return f"ERROR: {resp.status_code} - {resp.text}"
    
    return f"Created transaction ID {resp.json()['id']}"

Key insights:

  • Auth is separate. The server never sees credentials; a one-time login_setup.py script handles MFA and saves a session token. The MCP server reads the token from disk. This keeps secrets out of the process boundary.
  • Modularize by domain. With 40+ tools, organize them into logical groups (accounts, transactions, budgets, rules, tags). The agent can still call them all, but the code is navigable.
  • Dry-run for mutations. Any tool that writes includes a dry_run=True parameter. The agent can preview before committing.

Server 3: Inventory MCP

The problem: I run a small refurbishment side-project. I need Claude to query inventory, calculate P&L, and—carefully—update item statuses or log repairs. The data lives in a Postgres database behind the homelab.

The server: ~600 lines. ~9 read tools, 4 write tools, all behind a port-forward.

from fastmcp import FastMCP
import subprocess
import psycopg2
import json
import os
from datetime import datetime

app = FastMCP("inventory-mcp")

def get_db_connection():
    """Lazy port-forward to the database, reuse existing tunnel if open."""
    # Check if tunnel already exists
    try:
        conn = psycopg2.connect(
            host="localhost",
            port=5432,
            database="inventory",
            user="mcp_read"
        )
        return conn
    except:
        # Create new tunnel
        subprocess.Popen([
            "kubectl", "port-forward", "-n", "inventory",
            "svc/postgres", "5432:5432"
        ], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        time.sleep(2)
        return psycopg2.connect(
            host="localhost", port=5432, database="inventory", user="mcp_read"
        )

# Read tools — always available
@app.tool()
def get_portfolio_summary() -> str:
    """Summary: count, total cost, total value, gross margin %."""
    conn = get_db_connection()
    cur = conn.cursor()
    cur.execute("""
        SELECT
            COUNT(*) as count,
            COALESCE(SUM(cost), 0) as total_cost,
            COALESCE(SUM(current_value), 0) as total_value
        FROM inventory WHERE status != 'sold'
    """)
    row = cur.fetchone()
    conn.close()
    
    count, cost, value = row
    margin = ((value - cost) / cost * 100) if cost > 0 else 0
    return f"Portfolio: {count} items | Cost: ${cost:.2f} | Value: ${value:.2f} | Margin: {margin:.1f}%"

@app.tool()
def query_inventory(status: str | None = None, category: str | None = None, limit: int = 50) -> str:
    """List inventory items, optionally filtered by status or category."""
    conn = get_db_connection()
    cur = conn.cursor()
    
    query = "SELECT id, name, status, category, cost, current_value FROM inventory WHERE 1=1"
    params = []
    
    if status:
        query += " AND status = %s"
        params.append(status)
    if category:
        query += " AND category = %s"
        params.append(category)
    
    query += " LIMIT %s"
    params.append(limit)
    
    cur.execute(query, params)
    rows = cur.fetchall()
    conn.close()
    
    lines = []
    for row in rows:
        id_, name, st, cat, cost, value = row
        lines.append(f"- [{id_}] {name} ({cat}) - {st} | Cost: ${cost:.2f}, Value: ${value:.2f}")
    
    return "\n".join(lines) if lines else "No items found."

# Write tools — gated
@app.tool()
def update_item_status(item_id: int, new_status: str) -> str:
    """Update an item's status (e.g., 'in-repair' -> 'ready-to-sell'). Requires MCP_WRITE=1."""
    if not os.getenv("MCP_WRITE"):
        return "ERROR: Write mode disabled."
    
    if new_status not in ["in-stock", "in-repair", "ready-to-sell", "listed", "sold"]:
        return f"ERROR: Invalid status '{new_status}'."
    
    conn = get_db_connection()
    cur = conn.cursor()
    
    try:
        cur.execute(
            "UPDATE inventory SET status = %s, updated_at = %s WHERE id = %s",
            (new_status, datetime.utcnow(), item_id)
        )
        conn.commit()
        
        # Audit log
        cur.execute(
            "INSERT INTO audit_log (action, item_id, old_value, new_value, timestamp) VALUES (%s, %s, %s, %s, %s)",
            ("status_change", item_id, None, new_status, datetime.utcnow())
        )
        conn.commit()
    except Exception as e:
        return f"ERROR: {str(e)}"
    finally:
        conn.close()
    
    return f"Updated item {item_id} to '{new_status}'."

@app.tool()
def think(reasoning: str) -> str:
    """Scratchpad for the agent to reason before making a decision. Returns the input unchanged."""
    return f"Reasoning logged: {reasoning}"

if __name__ == "__main__":
    mcp.run(transport="stdio", app=app)

Key insights:

  • Reads always-on, writes gated. Every write tool checks os.getenv("MCP_WRITE"). In production, I only set MCP_WRITE=1 when I’m explicitly asking Claude to modify inventory.
  • Audit every mutation. Every write goes to an audit_log table with the item ID, old value, new value, timestamp, and action. I can replay decisions and debug issues.
  • Long operations are a problem. A 5-minute repair estimate blocks the stdio connection. I keep tools fast; long jobs go to a job queue outside the MCP boundary.
  • Lazy tunnels. The port-forward is created on first use and reused. Saves setup time and keeps the connection open for follow-up queries.
  • The think tool. It’s a no-op scratchpad. Before making a risky mutation, the agent can call think("Here's why I'm about to sell this item...") and I can review the reasoning in the logs.

Transport & return types

All three servers use stdio. FastMCP handles the JSON-RPC plumbing; you just define tools with @app.tool() and call mcp.run(transport="stdio").

Return types:

  • Success: Return a string or dict. The SDK serializes it to JSON.
  • Error: Return a string starting with "ERROR: ". The client sees it as a tool error and can retry or escalate.
# Good
return f"Found {count} pages"
return {"status": "ok", "items": []}

# Also good
return "ERROR: Invalid page ID"

# Bad — don't do this
raise Exception("...")  # Crashes the server
return None  # Ambiguous

Critical: Log to stderr, not stdout. The stdio transport uses stdout for JSON-RPC messages. Anything else breaks the protocol.

import sys
sys.stderr.write(f"DEBUG: {message}\n")  # OK
print(f"DEBUG: {message}")  # BREAKS THE SERVER

Auth & secrets

Three patterns I use:

1. Session tokens (finance server):

# One-time setup
python login_setup.py  # Prompts for password, does MFA, saves token to ~/.finance_session

# MCP server
SESSION = json.load(open(os.path.expanduser("~/.finance_session")))

2. Environment variables (inventory server):

export DATABASE_URL="postgres://..."
export MCP_WRITE=1

3. Kubectl secrets (if running in k8s):

kubectl create secret generic mcp-secrets --from-literal=api_token=xyz
# Mount to the pod and read at runtime

Rule: Never pass credentials through the MCP tool boundary. Load them at server startup. Never log them.

Schema drift

The biggest gotcha: if the upstream API adds or removes a field, the MCP tool contract breaks silently.

Example: Monarch Money adds a currency field to transactions. My tool doesn’t know about it. The agent asks for a transaction in USD, the API returns EUR, the tool ignores the mismatch, and the agent makes a decision based on wrong data.

Mitigation:

  • Keep docstrings and type hints in sync with the upstream API.
  • Add integration tests that call the real API and validate the response shape.
  • Log every tool call to a file; audit the logs for unexpected fields.
  • Set a reminder to test each server monthly.
@app.tool()
def list_transactions(account_id: str) -> str:
    """List recent transactions. Each has: date, payee, amount, category, currency."""
    # ^ Update this docstring whenever the API changes
    resp = requests.get(...)
    # Validate response shape
    for txn in resp.json()["data"]:
        assert "currency" in txn, "API schema changed!"
    ...

Three cross-cutting lessons

1. Flatten everything. If the upstream API is nested, unwrap it in the MCP tool. Return markdown for documents, simple dicts for data. Agents reason better with flat structures.

2. Modularize by domain, not by HTTP verb. Instead of POST /inventory and GET /inventory, write tools like query_inventory, create_transaction, update_item_status. The agent shouldn’t need to know REST conventions.

3. Audit is your insurance policy. Every write goes to a log. Every long operation has a dry_run preview. Every dangerous tool (delete, write) is gated by an environment variable. This lets you give Claude more autonomy without losing control.

These three servers handle 90% of my day-to-day automation. They’re boring, reliable, and they let Claude do useful work without needing to know SQL, GraphQL, or API shapes. If you’re building something similar, start with one domain, flatten the API, and audit everything you write.