Implementing Core Features
Build the server foundation and implement the four tools.
Assembling Your Modules
The capstone project brings together everything you have built in previous lessons. You will use your existing modules to create a complete Notes MCP Server.
Building on All Previous Lessons
This capstone imports from the modules you created in Lessons 3-9. Make sure all previous modules are in your project directory.
Your Complete Project Structure
text
your_project/
jsonrpc.py # Lesson 3
mcp_core.py # Lesson 4
tools.py # Lesson 5
resources.py # Lesson 6
prompts.py # Lesson 9
stdio_transport.py # Lesson 7
http_sse_transport.py # Lesson 8
notes_server.py # CAPSTONE - We'll create this
notes.json # Data storage (auto-created)How to Read Code Blocks
In this lesson:
- Filename in header = Write this code to that file
- "Add after X" = Append to your existing file
- bash blocks = Commands to run in your terminal
- "Expected Output" = What you should see (don't write this)
The Full Implementation
Below is the complete notes_server.py implementation. While this version
is self-contained for clarity, note that it follows the same patterns you learned:
- Tool registration patterns from Lesson 5
- Resource patterns from Lesson 6
- STDIO transport from Lesson 7
- Prompt patterns from Lesson 9
Section 1: Create notes_server.py
Create: notes_server.py
Create a new file called notes_server.py with the data model and storage:
python
notes_server.py
#!/usr/bin/env python3
"""
notes_server.py - A complete MCP server for managing notes.
This capstone project demonstrates tools, resources, and prompts
working together in a single MCP server.
"""
import json
import sys
import os
import uuid
from datetime import datetime, timezone
from dataclasses import dataclass, field
from typing import List, Optional, Dict, Any
# =============================================================================
# Data Model
# =============================================================================
@dataclass
class Note:
"""Represents a single note."""
id: str
title: str
content: str
tags: List[str] = field(default_factory=list)
created_at: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())
updated_at: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())
def to_dict(self) -> dict:
return {
"id": self.id,
"title": self.title,
"content": self.content,
"tags": self.tags,
"created_at": self.created_at,
"updated_at": self.updated_at,
}
@classmethod
def from_dict(cls, data: dict) -> "Note":
return cls(
id=data["id"],
title=data["title"],
content=data["content"],
tags=data.get("tags", []),
created_at=data.get("created_at", datetime.now(timezone.utc).isoformat()),
updated_at=data.get("updated_at", datetime.now(timezone.utc).isoformat()),
)
@staticmethod
def generate_id() -> str:
return str(uuid.uuid4())[:8]
# =============================================================================
# Storage
# =============================================================================
class NotesStorage:
"""Handle persistent storage of notes."""
def __init__(self, filepath: str = "notes.json"):
self.filepath = filepath
self.notes: List[Note] = []
self.load()
def load(self):
if os.path.exists(self.filepath):
try:
with open(self.filepath, "r") as f:
data = json.load(f)
self.notes = [Note.from_dict(n) for n in data]
except (json.JSONDecodeError, KeyError):
self.notes = []
else:
self.notes = []
def save(self):
with open(self.filepath, "w") as f:
json.dump([n.to_dict() for n in self.notes], f, indent=2)
def create(self, title: str, content: str, tags: Optional[List[str]] = None) -> Note:
note = Note(
id=Note.generate_id(),
title=title,
content=content,
tags=tags or [],
)
self.notes.append(note)
self.save()
return note
def get_by_id(self, note_id: str) -> Optional[Note]:
for note in self.notes:
if note.id == note_id:
return note
return None
def update(self, note_id: str, title: Optional[str] = None,
content: Optional[str] = None, tags: Optional[List[str]] = None) -> Optional[Note]:
note = self.get_by_id(note_id)
if not note:
return None
if title is not None:
note.title = title
if content is not None:
note.content = content
if tags is not None:
note.tags = tags
note.updated_at = datetime.now(timezone.utc).isoformat()
self.save()
return note
def delete(self, note_id: str) -> bool:
note = self.get_by_id(note_id)
if not note:
return False
self.notes.remove(note)
self.save()
return True
def search(self, query: Optional[str] = None, tag: Optional[str] = None) -> List[Note]:
results = self.notes
if tag:
results = [n for n in results if tag.lower() in [t.lower() for t in n.tags]]
if query:
query = query.lower()
results = [n for n in results
if query in n.title.lower() or query in n.content.lower()]
return results
def get_all_tags(self) -> List[str]:
tags = set()
for note in self.notes:
tags.update(note.tags)
return sorted(list(tags))Section 2: Server Class
Edit: notes_server.py
Add the NotesServer class after Section 1:
python
notes_server.py - Add after Section 1
# =============================================================================
# Section 2: MCP Server
# =============================================================================
class NotesServer:
"""MCP server for note management."""
PROTOCOL_VERSION = "2024-11-05"
def __init__(self, storage_path: str = "notes.json"):
self.storage = NotesStorage(storage_path)
self.initialized = False
def run(self):
"""Main loop: read from stdin, write to stdout."""
for line in sys.stdin:
line = line.strip()
if not line:
continue
try:
request = json.loads(line)
response = self.handle_request(request)
if response: # Don't respond to notifications
self.send_response(response)
except json.JSONDecodeError:
self.send_response(self.make_error(None, -32700, "Parse error"))
except Exception as e:
self.send_response(self.make_error(None, -32603, str(e)))
def send_response(self, response: dict):
"""Send a JSON-RPC response."""
print(json.dumps(response), flush=True)
def make_response(self, id: Any, result: Any) -> dict:
"""Create a successful JSON-RPC response."""
return {"jsonrpc": "2.0", "id": id, "result": result}
def make_error(self, id: Any, code: int, message: str, data: Any = None) -> dict:
"""Create a JSON-RPC error response."""
error = {"code": code, "message": message}
if data:
error["data"] = data
return {"jsonrpc": "2.0", "id": id, "error": error}
def handle_request(self, request: dict) -> Optional[dict]:
"""Route request to appropriate handler."""
method = request.get("method", "")
id = request.get("id")
params = request.get("params", {})
# Notifications (no id) don't get responses
if id is None and method == "notifications/initialized":
return None
handlers = {
"initialize": self.handle_initialize,
"tools/list": self.handle_tools_list,
"tools/call": self.handle_tools_call,
"resources/list": self.handle_resources_list,
"resources/read": self.handle_resources_read,
"prompts/list": self.handle_prompts_list,
"prompts/get": self.handle_prompts_get,
}
handler = handlers.get(method)
if handler:
try:
result = handler(params)
return self.make_response(id, result)
except Exception as e:
return self.make_error(id, -32603, f"Internal error: {e}")
else:
return self.make_error(id, -32601, f"Method not found: {method}")Section 3: Initialization Handler
Edit: notes_server.py
Add the initialization handler to the NotesServer class:
python
notes_server.py - Add to NotesServer class
# -------------------------------------------------------------------------
# Section 3: Initialization Handler
# -------------------------------------------------------------------------
def handle_initialize(self, params: dict) -> dict:
"""Handle the initialize request."""
self.initialized = True
return {
"protocolVersion": self.PROTOCOL_VERSION,
"capabilities": {
"tools": {},
"resources": {},
"prompts": {},
},
"serverInfo": {
"name": "notes-server",
"version": "1.0.0",
},
}Section 4: Tools List Handler
Edit: notes_server.py
Add the tools list handler after Section 3:
python
notes_server.py - Add after Section 3
# -------------------------------------------------------------------------
# Section 4: Tools List Handler
# -------------------------------------------------------------------------
def handle_tools_list(self, params: dict) -> dict:
"""Return list of available tools."""
return {
"tools": [
{
"name": "create_note",
"description": "Create a new note with title, content, and optional tags",
"inputSchema": {
"type": "object",
"properties": {
"title": {
"type": "string",
"description": "The title of the note"
},
"content": {
"type": "string",
"description": "The content/body of the note"
},
"tags": {
"type": "array",
"items": {"type": "string"},
"description": "Optional tags for categorization"
}
},
"required": ["title", "content"]
}
},
{
"name": "update_note",
"description": "Update an existing note by ID",
"inputSchema": {
"type": "object",
"properties": {
"id": {
"type": "string",
"description": "The ID of the note to update"
},
"title": {"type": "string", "description": "New title"},
"content": {"type": "string", "description": "New content"},
"tags": {
"type": "array",
"items": {"type": "string"},
"description": "New tags (replaces existing)"
}
},
"required": ["id"]
}
},
{
"name": "delete_note",
"description": "Delete a note by ID",
"inputSchema": {
"type": "object",
"properties": {
"id": {
"type": "string",
"description": "The ID of the note to delete"
}
},
"required": ["id"]
}
},
{
"name": "search_notes",
"description": "Search notes by keyword or tag",
"inputSchema": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Search keyword for title/content"
},
"tag": {
"type": "string",
"description": "Filter by tag"
}
}
}
}
]
}Section 5: Tools Call Handler
Edit: notes_server.py
Add the tools call handler and tool implementations after Section 4:
python
notes_server.py - Add after Section 4
# -------------------------------------------------------------------------
# Section 5: Tools Call Handler
# -------------------------------------------------------------------------
def handle_tools_call(self, params: dict) -> dict:
"""Execute a tool and return the result."""
name = params.get("name", "")
arguments = params.get("arguments", {})
tool_handlers = {
"create_note": self.tool_create_note,
"update_note": self.tool_update_note,
"delete_note": self.tool_delete_note,
"search_notes": self.tool_search_notes,
}
handler = tool_handlers.get(name)
if not handler:
return {
"content": [{"type": "text", "text": f"Unknown tool: {name}"}],
"isError": True,
}
try:
return handler(arguments)
except Exception as e:
return {
"content": [{"type": "text", "text": f"Error: {e}"}],
"isError": True,
}
# -------------------------------------------------------------------------
# Tool Implementations
# -------------------------------------------------------------------------
def tool_create_note(self, args: dict) -> dict:
"""Create a new note."""
title = args.get("title")
content = args.get("content")
tags = args.get("tags", [])
if not title or not content:
return {
"content": [{"type": "text", "text": "Error: title and content are required"}],
"isError": True,
}
note = self.storage.create(title, content, tags)
return {
"content": [{
"type": "text",
"text": f"Created note '{note.title}' with ID: {note.id}"
}]
}
def tool_update_note(self, args: dict) -> dict:
"""Update an existing note."""
note_id = args.get("id")
if not note_id:
return {
"content": [{"type": "text", "text": "Error: id is required"}],
"isError": True,
}
note = self.storage.update(
note_id,
title=args.get("title"),
content=args.get("content"),
tags=args.get("tags"),
)
if not note:
return {
"content": [{"type": "text", "text": f"Error: Note {note_id} not found"}],
"isError": True,
}
return {
"content": [{
"type": "text",
"text": f"Updated note '{note.title}' (ID: {note.id})"
}]
}
def tool_delete_note(self, args: dict) -> dict:
"""Delete a note."""
note_id = args.get("id")
if not note_id:
return {
"content": [{"type": "text", "text": "Error: id is required"}],
"isError": True,
}
if self.storage.delete(note_id):
return {
"content": [{"type": "text", "text": f"Deleted note {note_id}"}]
}
else:
return {
"content": [{"type": "text", "text": f"Error: Note {note_id} not found"}],
"isError": True,
}
def tool_search_notes(self, args: dict) -> dict:
"""Search notes by query or tag."""
query = args.get("query")
tag = args.get("tag")
results = self.storage.search(query=query, tag=tag)
if not results:
return {
"content": [{"type": "text", "text": "No notes found matching criteria"}]
}
text = f"Found {len(results)} note(s):\n\n"
for note in results:
tags_str = ", ".join(note.tags) if note.tags else "none"
text += f"- [{note.id}] {note.title} (tags: {tags_str})\n"
return {"content": [{"type": "text", "text": text}]}Tool Response Format
Tool responses must include a content array with content
items. Each item has a type (usually "text") and the
corresponding data. Set isError: true when returning errors.
Progress Check
At this point, you have implemented:
- The Note data model with serialization
- The NotesStorage class with CRUD operations
- The NotesServer class with STDIO transport
- The initialization handshake
- Four complete tools with error handling
Coming Up Next
In the next step, we will add resources and prompts to complete the server. Resources will expose notes as readable data, and prompts will help Claude work with notes more effectively.