Skip to content
This repository was archived by the owner on Mar 24, 2025. It is now read-only.

Commit 35730e4

Browse files
dsp-antdsp-ant
committedNov 24, 2024··
Move template to flat app pattern
·
v1.0.5
1 parent e5c6c63 commit 35730e4

File tree

2 files changed

+127
-152
lines changed

2 files changed

+127
-152
lines changed
 

‎src/create_mcp_server/template/server.py.jinja2

+126-151
Original file line numberDiff line numberDiff line change
@@ -6,173 +6,148 @@ from mcp.server import NotificationOptions, Server
66
from pydantic import AnyUrl
77
import mcp.server.stdio
88

9+
# Store notes as a simple key-value dict to demonstrate state management
10+
notes: dict[str, str] = {}
911

10-
class McpServer(Server):
12+
server = Server("{{server_name}}")
13+
14+
@server.list_resources()
15+
async def handle_list_resources() -> list[types.Resource]:
1116
"""
12-
Example MCP server demonstrating the three core primitives:
13-
1. Resources - Simple key-value notes storage accessed via custom URIs
14-
2. Prompts - Configurable summarization prompt for notes
15-
3. Tools - Ability to add new notes programmatically
16-
17-
This server shows how to:
18-
- Implement resource listing and reading with custom URI schemes
19-
- Create parameterized prompts that can access server state
20-
- Define tools with JSON schema validation
21-
- Send notifications when server state changes
17+
List available note resources.
18+
Each note is exposed as a resource with a custom note:// URI scheme.
2219
"""
20+
return [
21+
types.Resource(
22+
uri=AnyUrl(f"note://internal/{name}"),
23+
name=f"Note: {name}",
24+
description=f"A simple note named {name}",
25+
mimeType="text/plain",
26+
)
27+
for name in notes
28+
]
2329

24-
def __init__(self):
25-
super().__init__("{{server_name}}")
26-
27-
# Store notes as a simple key-value dict to demonstrate state management
28-
self.notes: dict[str, str] = {}
29-
30-
# RESOURCE HANDLERS
31-
# Resources allow clients to discover and read content from the server
32-
@self.list_resources()
33-
async def handle_list_resources() -> list[types.Resource]:
34-
"""
35-
List available note resources.
36-
Each note is exposed as a resource with a custom note:// URI scheme.
37-
"""
38-
return [
39-
types.Resource(
40-
uri=AnyUrl(f"note:///{name}"),
41-
name=f"Note: {name}",
42-
description=f"A simple note named {name}",
43-
mimeType="text/plain",
44-
)
45-
for name in self.notes
46-
]
47-
48-
@self.read_resource()
49-
async def handle_read_resource(uri: AnyUrl) -> str:
50-
"""
51-
Read a specific note's content by its URI.
52-
The note name is extracted from the URI host component.
53-
"""
54-
if uri.scheme != "note":
55-
raise ValueError(f"Unsupported URI scheme: {uri.scheme}")
56-
57-
name = uri.path
58-
if name is not None:
59-
name = name.lstrip("/")
60-
return self.notes[name]
61-
raise ValueError(f"Note not found: {name}")
62-
63-
# PROMPT HANDLERS
64-
# Prompts define templates that can be filled with server state
65-
@self.list_prompts()
66-
async def handle_list_prompts() -> list[types.Prompt]:
67-
"""
68-
List available prompts.
69-
Each prompt can have optional arguments to customize its behavior.
70-
"""
71-
return [
72-
types.Prompt(
73-
name="summarize-notes",
74-
description="Creates a summary of all notes",
75-
arguments=[
76-
types.PromptArgument(
77-
name="style",
78-
description="Style of the summary (brief/detailed)",
79-
required=False,
80-
)
81-
],
82-
)
83-
]
84-
85-
@self.get_prompt()
86-
async def handle_get_prompt(
87-
name: str, arguments: dict[str, str] | None
88-
) -> types.GetPromptResult:
89-
"""
90-
Generate a prompt by combining arguments with server state.
91-
The prompt includes all current notes and can be customized via arguments.
92-
"""
93-
if name != "summarize-notes":
94-
raise ValueError(f"Unknown prompt: {name}")
95-
96-
style = (arguments or {}).get("style", "brief")
97-
detail_prompt = " Give extensive details." if style == "detailed" else ""
98-
99-
return types.GetPromptResult(
100-
description="Summarize the current notes",
101-
messages=[
102-
types.PromptMessage(
103-
role="user",
104-
content=types.TextContent(
105-
type="text",
106-
text=f"Here are the current notes to summarize:{detail_prompt}\n\n"
107-
+ "\n".join(
108-
f"- {name}: {content}"
109-
for name, content in self.notes.items()
110-
),
111-
),
112-
)
113-
],
114-
)
30+
@server.read_resource()
31+
async def handle_read_resource(uri: AnyUrl) -> str:
32+
"""
33+
Read a specific note's content by its URI.
34+
The note name is extracted from the URI host component.
35+
"""
36+
if uri.scheme != "note":
37+
raise ValueError(f"Unsupported URI scheme: {uri.scheme}")
11538

116-
# TOOL HANDLERS
117-
# Tools allow clients to modify server state in controlled ways
118-
@self.list_tools()
119-
async def handle_list_tools() -> list[types.Tool]:
120-
"""
121-
List available tools.
122-
Each tool specifies its arguments using JSON Schema validation.
123-
"""
124-
return [
125-
types.Tool(
126-
name="add-note",
127-
description="Add a new note",
128-
inputSchema={
129-
"type": "object",
130-
"properties": {
131-
"name": {"type": "string"},
132-
"content": {"type": "string"},
133-
},
134-
"required": ["name", "content"],
135-
},
39+
name = uri.path
40+
if name is not None:
41+
name = name.lstrip("/")
42+
return notes[name]
43+
raise ValueError(f"Note not found: {name}")
44+
45+
@server.list_prompts()
46+
async def handle_list_prompts() -> list[types.Prompt]:
47+
"""
48+
List available prompts.
49+
Each prompt can have optional arguments to customize its behavior.
50+
"""
51+
return [
52+
types.Prompt(
53+
name="summarize-notes",
54+
description="Creates a summary of all notes",
55+
arguments=[
56+
types.PromptArgument(
57+
name="style",
58+
description="Style of the summary (brief/detailed)",
59+
required=False,
13660
)
137-
]
61+
],
62+
)
63+
]
64+
65+
@server.get_prompt()
66+
async def handle_get_prompt(
67+
name: str, arguments: dict[str, str] | None
68+
) -> types.GetPromptResult:
69+
"""
70+
Generate a prompt by combining arguments with server state.
71+
The prompt includes all current notes and can be customized via arguments.
72+
"""
73+
if name != "summarize-notes":
74+
raise ValueError(f"Unknown prompt: {name}")
75+
76+
style = (arguments or {}).get("style", "brief")
77+
detail_prompt = " Give extensive details." if style == "detailed" else ""
78+
79+
return types.GetPromptResult(
80+
description="Summarize the current notes",
81+
messages=[
82+
types.PromptMessage(
83+
role="user",
84+
content=types.TextContent(
85+
type="text",
86+
text=f"Here are the current notes to summarize:{detail_prompt}\n\n"
87+
+ "\n".join(
88+
f"- {name}: {content}"
89+
for name, content in notes.items()
90+
),
91+
),
92+
)
93+
],
94+
)
13895

139-
@self.call_tool()
140-
async def handle_call_tool(
141-
name: str, arguments: dict | None
142-
) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
143-
"""
144-
Handle tool execution requests.
145-
Tools can modify server state and notify clients of changes.
146-
"""
147-
if name != "add-note":
148-
raise ValueError(f"Unknown tool: {name}")
96+
@server.list_tools()
97+
async def handle_list_tools() -> list[types.Tool]:
98+
"""
99+
List available tools.
100+
Each tool specifies its arguments using JSON Schema validation.
101+
"""
102+
return [
103+
types.Tool(
104+
name="add-note",
105+
description="Add a new note",
106+
inputSchema={
107+
"type": "object",
108+
"properties": {
109+
"name": {"type": "string"},
110+
"content": {"type": "string"},
111+
},
112+
"required": ["name", "content"],
113+
},
114+
)
115+
]
149116

150-
if not arguments:
151-
raise ValueError("Missing arguments")
117+
@server.call_tool()
118+
async def handle_call_tool(
119+
name: str, arguments: dict | None
120+
) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
121+
"""
122+
Handle tool execution requests.
123+
Tools can modify server state and notify clients of changes.
124+
"""
125+
if name != "add-note":
126+
raise ValueError(f"Unknown tool: {name}")
152127

153-
note_name = arguments.get("name")
154-
content = arguments.get("content")
128+
if not arguments:
129+
raise ValueError("Missing arguments")
155130

156-
if not note_name or not content:
157-
raise ValueError("Missing name or content")
131+
note_name = arguments.get("name")
132+
content = arguments.get("content")
158133

159-
# Update server state
160-
self.notes[note_name] = content
134+
if not note_name or not content:
135+
raise ValueError("Missing name or content")
161136

162-
# Notify clients that resources have changed
163-
await self.request_context.session.send_resource_list_changed()
137+
# Update server state
138+
notes[note_name] = content
164139

165-
return [
166-
types.TextContent(
167-
type="text",
168-
text=f"Added note '{note_name}' with content: {content}",
169-
)
170-
]
140+
# Notify clients that resources have changed
141+
await server.request_context.session.send_resource_list_changed()
171142

143+
return [
144+
types.TextContent(
145+
type="text",
146+
text=f"Added note '{note_name}' with content: {content}",
147+
)
148+
]
172149

173150
async def main():
174-
server = McpServer()
175-
176151
# Run the server using stdin/stdout streams
177152
async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
178153
await server.run(

‎uv.lock

+1-1
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

0 commit comments

Comments
 (0)
This repository has been archived.