Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly
Sign up
Appearance settings

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)

AltStyle によって変換されたページ (->オリジナル) /