Build your own MCP server with FastMCP — connect a LLM to your business tools in Python

mkdir roger-mcp && cd roger-mcp
python -m venv .venv && source .venv/bin/activate  # Windows: .venv\Scripts\activate
pip install fastmcp

from fastmcp import FastMCP

mcp = FastMCP("roger")

@mcp.tool()
def ping() -> str:
    """Sanity check tool: returns 'pong' if the server is alive."""
    return "pong"

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

python server.py

from fastmcp import FastMCP
from pydantic import BaseModel, Field
from typing import Literal
import httpx, os

NOTION_TOKEN = os.environ["NOTION_TOKEN"]
NOTION_DB_TASKS = os.environ["NOTION_DB_TASKS"]

mcp = FastMCP("roger")
client = httpx.Client(
    base_url="https://api.notion.com/v1",
    headers={
        "Authorization": f"Bearer {NOTION_TOKEN}",
        "Notion-Version": "2022-06-28",
    },
    timeout=15.0,
)

class TaskInput(BaseModel):
    title: str = Field(..., min_length=1, max_length=200)
    status: Literal["todo", "doing", "done"] = "todo"
    project: str | None = None
    due: str | None = Field(default=None, description="ISO 8601 date (YYYY-MM-DD)")

@mcp.tool()
def search_notion(query: str, limit: int = 10) -> list[dict]:
    """Search across all Notion pages the integration has access to. Use for open-ended lookups where you don't know the database in advance."""
    r = client.post("/search", json={"query": query, "page_size": limit})
    r.raise_for_status()
    return [{"id": p["id"], "title": _title(p), "url": p["url"]} for p in r.json()["results"]]

@mcp.tool()
def upsert_task(task: TaskInput) -> dict:
    """Create or update a task in the internal task database. Use when the user asks to add, plan or reschedule work."""
    payload = {"parent": {"database_id": NOTION_DB_TASKS}, "properties": _task_to_props(task)}
    r = client.post("/pages", json=payload)
    r.raise_for_status()
    return {"id": r.json()["id"], "url": r.json()["url"], "status": task.status}

@mcp.tool()
def list_projects(status: Literal["active", "archived", "all"] = "active") -> list[dict]:
    """List all internal projects with their current status. Use to discover which project a new task belongs to."""
    r = client.post(f"/databases/{os.environ['NOTION_DB_PROJECTS']}/query")
    r.raise_for_status()
    return [{"id": p["id"], "name": _title(p), "status": _select(p, "Status")} for p in r.json()["results"]
            if status == "all" or _select(p, "Status") == status]

@mcp.tool()
def summarize_page(page_id: str) -> str:
    """Fetch the raw markdown of a Notion page. Use when the user asks about the content of a specific page you already found via search_notion."""
    r = client.get(f"/blocks/{page_id}/children")
    r.raise_for_status()
    return _blocks_to_markdown(r.json()["results"])

@mcp.tool()
def dispatch_email(to: str, subject: str, body_markdown: str) -> dict:
    """Send an email via the internal transactional mailer. Use only when the user explicitly asks to email someone — never proactively."""
    # Real implementation: Postmark, Resend, or a local SMTP relay. Here: stubbed.
    return {"queued": True, "to": to, "subject": subject, "size": len(body_markdown)}

@mcp.tool()
def get_metrics(period: Literal["day", "week", "month"] = "week") -> dict:
    """Return internal KPIs (MRR, active clients, open tasks) for the given period. Read-only. Cached 60 s."""
    return {"period": period, "mrr_eur": 0, "active_clients": 0, "open_tasks": 0}

# Helpers _title, _select, _task_to_props, _blocks_to_markdown intentionally omitted for readability.

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

import time
from functools import wraps

_cache: dict[tuple, tuple[float, object]] = {}

def cached(ttl: int = 60):
    def deco(fn):
        @wraps(fn)
        def wrapper(*args, **kwargs):
            key = (fn.__name__, args, tuple(sorted(kwargs.items())))
            hit = _cache.get(key)
            if hit and time.time() - hit[0] < ttl:
                return hit[1]
            result = fn(*args, **kwargs)
            _cache[key] = (time.time(), result)
            return result
        return wrapper
    return deco

@mcp.tool()
@cached(ttl=60)
def list_projects(status: Literal["active", "archived", "all"] = "active") -> list[dict]:
    ...

@mcp.tool()
@cached(ttl=60)
def search_notion(query: str, limit: int = 10) -> list[dict]:
    ...

import random, time
import httpx

def with_backoff(fn, max_retries: int = 3):
    for attempt in range(max_retries):
        try:
            return fn()
        except httpx.HTTPStatusError as e:
            if e.response.status_code not in (429, 502, 503, 504):
                raise
            if attempt == max_retries - 1:
                raise
            sleep_s = (2 ** attempt) + random.random()  # 1-2s, 2-3s, 4-5s
            time.sleep(sleep_s)

class Circuit:
    def __init__(self, cool_down: int = 60):
        self.failures = 0
        self.opened_at: float | None = None
        self.cool_down = cool_down

    def check(self):
        if self.opened_at and time.time() - self.opened_at < self.cool_down:
            raise RuntimeError("Notion currently degraded — try again shortly.")
        if self.opened_at and time.time() - self.opened_at >= self.cool_down:
            self.opened_at = None
            self.failures = 0

    def record_failure(self):
        self.failures += 1
        if self.failures >= 5:
            self.opened_at = time.time()

notion_circuit = Circuit()

# stdio (default)
mcp.run()

# HTTP on port 8000
mcp.run(transport="http", host="0.0.0.0", port=8000)

from fastmcp.server.auth import BearerAuth

expected_token = os.environ["MCP_BEARER_TOKEN"]

mcp = FastMCP("roger", auth=BearerAuth(token=expected_token))

# /etc/systemd/system/roger-mcp.service
[Unit]
Description=Roger MCP server
After=network.target

[Service]
Type=simple
User=roger
WorkingDirectory=/opt/roger-mcp
EnvironmentFile=/etc/roger-mcp.env  # NOTION_TOKEN, MCP_BEARER_TOKEN, etc.
ExecStart=/opt/roger-mcp/.venv/bin/python server.py
Restart=on-failure
RestartSec=5

# Hardening
NoNewPrivileges=true
PrivateTmp=true
ProtectSystem=strict
ProtectHome=true
ReadWritePaths=/opt/roger-mcp/data
CapabilityBoundingSet=

[Install]
WantedBy=multi-user.target

sudo systemctl enable --now roger-mcp
sudo systemctl status roger-mcp

mcp.mag-cie.com {
  reverse_proxy 127.0.0.1:8000
}

{
  "mcpServers": {
    "roger": {
      "command": "/opt/roger-mcp/.venv/bin/python",
      "args": ["/opt/roger-mcp/server.py"],
      "env": {
        "NOTION_TOKEN": "secret_...",
        "NOTION_DB_TASKS": "..."
      }
    }
  }
}

{
  "mcpServers": {
    "roger-remote": {
      "command": "mcp-proxy",
      "args": [
        "https://mcp.mag-cie.com/",
        "--headers", "Authorization=Bearer eyJhbGc..."
      ]
    }
  }
}

hermes mcp add roger \
  --url https://mcp.mag-cie.com/ \
  --header "Authorization: Bearer eyJhbGc..."

hermes mcp list   # should list roger and its 6 tools

from mcp import ClientSession
from mcp.client.stdio import stdio_client, StdioServerParameters
import asyncio

async def main():
    params = StdioServerParameters(command="python", args=["server.py"])
    async with stdio_client(params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            tools = await session.list_tools()
            assert {t.name for t in tools.tools} >= {
                "ping", "search_notion", "upsert_task",
                "list_projects", "summarize_page", "get_metrics",
            }
            r = await session.call_tool("ping", {})
            assert r.content[0].text == "pong"
            print("OK — MCP server ready")

asyncio.run(main())