Build an MCP Server in Python with FastMCP

# todo_server.py
from typing import Literal
from itertools import count
from datetime import datetime, timezone
from fastmcp import FastMCP

# In-memory storage for demo purposes
TODOS: list[dict] = []
_id = count(start=1)

mcp = FastMCP(name="Todo Manager")

@mcp.tool
def create_todo(
    title: str,
    description: str = "",
    priority: Literal["low", "medium", "high"] = "medium",
) -> dict:
    """Create a todo (id, title, status, priority, timestamps)."""
    todo = {
        "id": next(_id),
        "title": title,
        "description": description,
        "priority": priority,
        "status": "open",
        "created_at": datetime.now(timezone.utc).isoformat(),
        "completed_at": None,
    }
    TODOS.append(todo)
    return todo

@mcp.tool
def list_todos(status: Literal["open", "done", "all"] = "open") -> dict:
    """List todos by status ('open' | 'done' | 'all')."""
    if status == "all":
        items = TODOS
    elif status == "open":
        items = [t for t in TODOS if t["status"] == "open"]
    else:
        items = [t for t in TODOS if t["status"] == "done"]
    # Wrap arrays in a dict for clean JSON serialization in clients
    return {"items": items}

@mcp.tool
def complete_todo(todo_id: int) -> dict:
    """Mark a todo as done."""
    for t in TODOS:
        if t["id"] == todo_id:
            t["status"] = "done"
            t["completed_at"] = datetime.now(timezone.utc).isoformat()
            return t
    raise ValueError(f"Todo {todo_id} not found")

@mcp.tool
def search_todos(query: str) -> dict:
    """Case-insensitive search in title/description."""
    q = query.lower().strip()
    items = [t for t in TODOS if q in t["title"].lower() or q in t["description"].lower()]
    return {"items": items}

# Read-only resources
@mcp.resource("stats://todos")
def todo_stats() -> dict:
    """Aggregated stats: total, open, done."""
    total = len(TODOS)
    open_count = sum(1 for t in TODOS if t["status"] == "open")
    done_count = total - open_count
    return {"total": total, "open": open_count, "done": done_count}

@mcp.resource("todo://{id}")
def get_todo(id: int) -> dict:
    """Fetch a single todo by id."""
    for t in TODOS:
        if t["id"] == id:
            return t
    raise ValueError(f"Todo {id} not found")

# A reusable prompt
@mcp.prompt
def suggest_next_action(pending: int, project: str | None = None) -> str:
    """Render a small instruction for an LLM to propose next action."""
    base = f"You have {pending} pending TODOs. "
    if project:
        base += f"They relate to the project '{project}'. "
    base += "Suggest the most impactful next action in one short sentence."
    return base

if __name__ == "__main__":
    # Default transport is stdio; you can also use transport="http", host=..., port=...
    mcp.run()

python todo_server.py   # stdio transport (default)

# Replace the last line with:
mcp.run(transport="http", host="127.0.0.1", port=8000)

# todo_client_test.py
import asyncio
from fastmcp import Client

async def main():
    # Option A: Connect to local Python script (stdio)
    client = Client("todo_server.py")

    # Option B: In-memory (for tests)
    # from todo_server import mcp
    # client = Client(mcp)

    async with client:
        await client.ping()
        print("[OK] Connected")

        # Create a few todos
        t1 = await client.call_tool("create_todo", {"title": "Write README", "priority": "high"})
        t2 = await client.call_tool("create_todo", {"title": "Refactor utils", "description": "Split helpers into modules"})
        t3 = await client.call_tool("create_todo", {"title": "Add tests", "priority": "low"})
        print("Created IDs:", t1.data["id"], t2.data["id"], t3.data["id"])

        # List open
        open_list = await client.call_tool("list_todos", {"status": "open"})
        print("Open IDs:", [t["id"] for t in open_list.data["items"]])

        # Complete one
        updated = await client.call_tool("complete_todo", {"todo_id": t2.data["id"]})
        print("Completed:", updated.data["id"], "status:", updated.data["status"])

        # Search
        found = await client.call_tool("search_todos", {"query": "readme"})
        print("Search 'readme':", [t["id"] for t in found.data["items"]])

        # Resources
        stats = await client.read_resource("stats://todos")
        print("Stats:", getattr(stats[0], "text", None) or stats[0])

        todo2 = await client.read_resource(f"todo://{t2.data['id']}")
        print("todo://{id}:", getattr(todo2[0], "text", None) or todo2[0])

        # Prompt
        prompt_msgs = await client.get_prompt("suggest_next_action", {"pending": 2, "project": "MCP tutorial"})
        msgs_pretty = [
            {"role": m.role, "content": getattr(m, "content", None) or getattr(m, "text", None)}
            for m in getattr(prompt_msgs, "messages", [])
        ]
        print("Prompt messages:", msgs_pretty)

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

pip install fastmcp
python todo_client_test.py

{
  "mcpServers": {
    "todo-manager": {
      "command": "python",
      "args": ["todo_server.py"],
      "cwd": "/absolute/path/to/your/project",
      "env": {}
    }
  }
}

{
  "mcpServers": {
    "todo-manager": {
      "command": "python",
      "args": ["todo_server.py"],
      "cwd": "/absolute/path/to/your/project"
    }
  }
}

{
  "mcpServers": {
    "todo-manager": {
      "command": "python",
      "args": ["todo_server.py"],
      "cwd": "/absolute/path/to/your/project",
      "env": {}
    }
  }
}

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