Skip to content

Commit 2b6507f

Browse files
committed
Add FastMCP MCP Todo Manager example (server, client, README, requirements) under general/fastmcp-mcp-client-server-todo-manager.
1 parent e085c5f commit 2b6507f

File tree

4 files changed

+178
-0
lines changed

4 files changed

+178
-0
lines changed
Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,39 @@
1+
# Build a real MCP client and server in Python with FastMCP (Todo Manager example)
2+
3+
This folder contains the code that accompanies the article:
4+
5+
- Article: https://www.thepythoncode.com/article/fastmcp-mcp-client-server-todo-manager
6+
7+
What’s included
8+
- `todo_server.py`: FastMCP MCP server exposing tools, resources, and a prompt for a Todo Manager.
9+
- `todo_client_test.py`: A small client script that connects to the server and exercises all features.
10+
- `requirements.txt`: Python dependencies for this tutorial.
11+
12+
Quick start
13+
1) Install requirements
14+
```bash
15+
python -m venv .venv && source .venv/bin/activate # or use your preferred env manager
16+
pip install -r requirements.txt
17+
```
18+
19+
2) Run the server (stdio transport by default)
20+
```bash
21+
python todo_server.py
22+
```
23+
24+
3) In a separate terminal, run the client
25+
```bash
26+
python todo_client_test.py
27+
```
28+
29+
Optional: run the server over HTTP
30+
- In `todo_server.py`, replace the last line with:
31+
```python
32+
mcp.run(transport="http", host="127.0.0.1", port=8000)
33+
```
34+
- Then change the client constructor to `Client("http://127.0.0.1:8000/mcp")`.
35+
36+
Notes
37+
- Requires Python 3.10+.
38+
- The example uses in-memory storage for simplicity.
39+
- For production tips (HTTPS, auth, containerization), see the article.
Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
fastmcp>=2.12
Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,50 @@
1+
import asyncio
2+
from fastmcp import Client
3+
4+
async def main():
5+
# Option A: Connect to local Python script (stdio)
6+
client = Client("todo_server.py")
7+
8+
# Option B: In-memory (for tests)
9+
# from todo_server import mcp
10+
# client = Client(mcp)
11+
12+
async with client:
13+
await client.ping()
14+
print("[OK] Connected")
15+
16+
# Create a few todos
17+
t1 = await client.call_tool("create_todo", {"title": "Write README", "priority": "high"})
18+
t2 = await client.call_tool("create_todo", {"title": "Refactor utils", "description": "Split helpers into modules"})
19+
t3 = await client.call_tool("create_todo", {"title": "Add tests", "priority": "low"})
20+
print("Created IDs:", t1.data["id"], t2.data["id"], t3.data["id"])
21+
22+
# List open
23+
open_list = await client.call_tool("list_todos", {"status": "open"})
24+
print("Open IDs:", [t["id"] for t in open_list.data["items"]])
25+
26+
# Complete one
27+
updated = await client.call_tool("complete_todo", {"todo_id": t2.data["id"]})
28+
print("Completed:", updated.data["id"], "status:", updated.data["status"])
29+
30+
# Search
31+
found = await client.call_tool("search_todos", {"query": "readme"})
32+
print("Search 'readme':", [t["id"] for t in found.data["items"]])
33+
34+
# Resources
35+
stats = await client.read_resource("stats://todos")
36+
print("Stats:", getattr(stats[0], "text", None) or stats[0])
37+
38+
todo2 = await client.read_resource(f"todo://{t2.data['id']}")
39+
print("todo://{id}:", getattr(todo2[0], "text", None) or todo2[0])
40+
41+
# Prompt
42+
prompt_msgs = await client.get_prompt("suggest_next_action", {"pending": 2, "project": "MCP tutorial"})
43+
msgs_pretty = [
44+
{"role": m.role, "content": getattr(m, "content", None) or getattr(m, "text", None)}
45+
for m in getattr(prompt_msgs, "messages", [])
46+
]
47+
print("Prompt messages:", msgs_pretty)
48+
49+
if __name__ == "__main__":
50+
asyncio.run(main())
Lines changed: 88 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,88 @@
1+
from typing import Literal
2+
from itertools import count
3+
from datetime import datetime, timezone
4+
from fastmcp import FastMCP
5+
6+
# In-memory storage for demo purposes
7+
TODOS: list[dict] = []
8+
_id = count(start=1)
9+
10+
mcp = FastMCP(name="Todo Manager")
11+
12+
@mcp.tool
13+
def create_todo(
14+
title: str,
15+
description: str = "",
16+
priority: Literal["low", "medium", "high"] = "medium",
17+
) -> dict:
18+
"""Create a todo (id, title, status, priority, timestamps)."""
19+
todo = {
20+
"id": next(_id),
21+
"title": title,
22+
"description": description,
23+
"priority": priority,
24+
"status": "open",
25+
"created_at": datetime.now(timezone.utc).isoformat(),
26+
"completed_at": None,
27+
}
28+
TODOS.append(todo)
29+
return todo
30+
31+
@mcp.tool
32+
def list_todos(status: Literal["open", "done", "all"] = "open") -> dict:
33+
"""List todos by status ('open' | 'done' | 'all')."""
34+
if status == "all":
35+
items = TODOS
36+
elif status == "open":
37+
items = [t for t in TODOS if t["status"] == "open"]
38+
else:
39+
items = [t for t in TODOS if t["status"] == "done"]
40+
return {"items": items}
41+
42+
@mcp.tool
43+
def complete_todo(todo_id: int) -> dict:
44+
"""Mark a todo as done."""
45+
for t in TODOS:
46+
if t["id"] == todo_id:
47+
t["status"] = "done"
48+
t["completed_at"] = datetime.now(timezone.utc).isoformat()
49+
return t
50+
raise ValueError(f"Todo {todo_id} not found")
51+
52+
@mcp.tool
53+
def search_todos(query: str) -> dict:
54+
"""Case-insensitive search in title/description."""
55+
q = query.lower().strip()
56+
items = [t for t in TODOS if q in t["title"].lower() or q in t["description"].lower()]
57+
return {"items": items}
58+
59+
# Read-only resources
60+
@mcp.resource("stats://todos")
61+
def todo_stats() -> dict:
62+
"""Aggregated stats: total, open, done."""
63+
total = len(TODOS)
64+
open_count = sum(1 for t in TODOS if t["status"] == "open")
65+
done_count = total - open_count
66+
return {"total": total, "open": open_count, "done": done_count}
67+
68+
@mcp.resource("todo://{id}")
69+
def get_todo(id: int) -> dict:
70+
"""Fetch a single todo by id."""
71+
for t in TODOS:
72+
if t["id"] == id:
73+
return t
74+
raise ValueError(f"Todo {id} not found")
75+
76+
# A reusable prompt
77+
@mcp.prompt
78+
def suggest_next_action(pending: int, project: str | None = None) -> str:
79+
"""Render a small instruction for an LLM to propose next action."""
80+
base = f"You have {pending} pending TODOs. "
81+
if project:
82+
base += f"They relate to the project '{project}'. "
83+
base += "Suggest the most impactful next action in one short sentence."
84+
return base
85+
86+
if __name__ == "__main__":
87+
# Default transport is stdio; you can also use transport="http", host=..., port=...
88+
mcp.run()

0 commit comments

Comments
 (0)