View on Github
todo_server.py
# 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()
todo_client_test.py
# 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())