Move template to flat app pattern
This commit is contained in:
David Soria Parra
parent
e5c6c63923
commit
35730e4a16
2 changed files with 127 additions and 152 deletions
|
|
@ -6,173 +6,148 @@ from mcp.server import NotificationOptions, Server
|
|||
from pydantic import AnyUrl
|
||||
import mcp.server.stdio
|
||||
|
||||
# Store notes as a simple key-value dict to demonstrate state management
|
||||
notes: dict[str, str] = {}
|
||||
|
||||
class McpServer(Server):
|
||||
server = Server("{{server_name}}")
|
||||
|
||||
@server.list_resources()
|
||||
async def handle_list_resources() -> list[types.Resource]:
|
||||
"""
|
||||
Example MCP server demonstrating the three core primitives:
|
||||
1. Resources - Simple key-value notes storage accessed via custom URIs
|
||||
2. Prompts - Configurable summarization prompt for notes
|
||||
3. Tools - Ability to add new notes programmatically
|
||||
|
||||
This server shows how to:
|
||||
- Implement resource listing and reading with custom URI schemes
|
||||
- Create parameterized prompts that can access server state
|
||||
- Define tools with JSON schema validation
|
||||
- Send notifications when server state changes
|
||||
List available note resources.
|
||||
Each note is exposed as a resource with a custom note:// URI scheme.
|
||||
"""
|
||||
return [
|
||||
types.Resource(
|
||||
uri=AnyUrl(f"note://internal/{name}"),
|
||||
name=f"Note: {name}",
|
||||
description=f"A simple note named {name}",
|
||||
mimeType="text/plain",
|
||||
)
|
||||
for name in notes
|
||||
]
|
||||
|
||||
def __init__(self):
|
||||
super().__init__("{{server_name}}")
|
||||
@server.read_resource()
|
||||
async def handle_read_resource(uri: AnyUrl) -> str:
|
||||
"""
|
||||
Read a specific note's content by its URI.
|
||||
The note name is extracted from the URI host component.
|
||||
"""
|
||||
if uri.scheme != "note":
|
||||
raise ValueError(f"Unsupported URI scheme: {uri.scheme}")
|
||||
|
||||
# Store notes as a simple key-value dict to demonstrate state management
|
||||
self.notes: dict[str, str] = {}
|
||||
name = uri.path
|
||||
if name is not None:
|
||||
name = name.lstrip("/")
|
||||
return notes[name]
|
||||
raise ValueError(f"Note not found: {name}")
|
||||
|
||||
# RESOURCE HANDLERS
|
||||
# Resources allow clients to discover and read content from the server
|
||||
@self.list_resources()
|
||||
async def handle_list_resources() -> list[types.Resource]:
|
||||
"""
|
||||
List available note resources.
|
||||
Each note is exposed as a resource with a custom note:// URI scheme.
|
||||
"""
|
||||
return [
|
||||
types.Resource(
|
||||
uri=AnyUrl(f"note:///{name}"),
|
||||
name=f"Note: {name}",
|
||||
description=f"A simple note named {name}",
|
||||
mimeType="text/plain",
|
||||
@server.list_prompts()
|
||||
async def handle_list_prompts() -> list[types.Prompt]:
|
||||
"""
|
||||
List available prompts.
|
||||
Each prompt can have optional arguments to customize its behavior.
|
||||
"""
|
||||
return [
|
||||
types.Prompt(
|
||||
name="summarize-notes",
|
||||
description="Creates a summary of all notes",
|
||||
arguments=[
|
||||
types.PromptArgument(
|
||||
name="style",
|
||||
description="Style of the summary (brief/detailed)",
|
||||
required=False,
|
||||
)
|
||||
for name in self.notes
|
||||
]
|
||||
],
|
||||
)
|
||||
]
|
||||
|
||||
@self.read_resource()
|
||||
async def handle_read_resource(uri: AnyUrl) -> str:
|
||||
"""
|
||||
Read a specific note's content by its URI.
|
||||
The note name is extracted from the URI host component.
|
||||
"""
|
||||
if uri.scheme != "note":
|
||||
raise ValueError(f"Unsupported URI scheme: {uri.scheme}")
|
||||
@server.get_prompt()
|
||||
async def handle_get_prompt(
|
||||
name: str, arguments: dict[str, str] | None
|
||||
) -> types.GetPromptResult:
|
||||
"""
|
||||
Generate a prompt by combining arguments with server state.
|
||||
The prompt includes all current notes and can be customized via arguments.
|
||||
"""
|
||||
if name != "summarize-notes":
|
||||
raise ValueError(f"Unknown prompt: {name}")
|
||||
|
||||
name = uri.path
|
||||
if name is not None:
|
||||
name = name.lstrip("/")
|
||||
return self.notes[name]
|
||||
raise ValueError(f"Note not found: {name}")
|
||||
style = (arguments or {}).get("style", "brief")
|
||||
detail_prompt = " Give extensive details." if style == "detailed" else ""
|
||||
|
||||
# PROMPT HANDLERS
|
||||
# Prompts define templates that can be filled with server state
|
||||
@self.list_prompts()
|
||||
async def handle_list_prompts() -> list[types.Prompt]:
|
||||
"""
|
||||
List available prompts.
|
||||
Each prompt can have optional arguments to customize its behavior.
|
||||
"""
|
||||
return [
|
||||
types.Prompt(
|
||||
name="summarize-notes",
|
||||
description="Creates a summary of all notes",
|
||||
arguments=[
|
||||
types.PromptArgument(
|
||||
name="style",
|
||||
description="Style of the summary (brief/detailed)",
|
||||
required=False,
|
||||
)
|
||||
],
|
||||
)
|
||||
]
|
||||
|
||||
@self.get_prompt()
|
||||
async def handle_get_prompt(
|
||||
name: str, arguments: dict[str, str] | None
|
||||
) -> types.GetPromptResult:
|
||||
"""
|
||||
Generate a prompt by combining arguments with server state.
|
||||
The prompt includes all current notes and can be customized via arguments.
|
||||
"""
|
||||
if name != "summarize-notes":
|
||||
raise ValueError(f"Unknown prompt: {name}")
|
||||
|
||||
style = (arguments or {}).get("style", "brief")
|
||||
detail_prompt = " Give extensive details." if style == "detailed" else ""
|
||||
|
||||
return types.GetPromptResult(
|
||||
description="Summarize the current notes",
|
||||
messages=[
|
||||
types.PromptMessage(
|
||||
role="user",
|
||||
content=types.TextContent(
|
||||
type="text",
|
||||
text=f"Here are the current notes to summarize:{detail_prompt}\n\n"
|
||||
+ "\n".join(
|
||||
f"- {name}: {content}"
|
||||
for name, content in self.notes.items()
|
||||
),
|
||||
),
|
||||
)
|
||||
],
|
||||
)
|
||||
|
||||
# TOOL HANDLERS
|
||||
# Tools allow clients to modify server state in controlled ways
|
||||
@self.list_tools()
|
||||
async def handle_list_tools() -> list[types.Tool]:
|
||||
"""
|
||||
List available tools.
|
||||
Each tool specifies its arguments using JSON Schema validation.
|
||||
"""
|
||||
return [
|
||||
types.Tool(
|
||||
name="add-note",
|
||||
description="Add a new note",
|
||||
inputSchema={
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"name": {"type": "string"},
|
||||
"content": {"type": "string"},
|
||||
},
|
||||
"required": ["name", "content"],
|
||||
},
|
||||
)
|
||||
]
|
||||
|
||||
@self.call_tool()
|
||||
async def handle_call_tool(
|
||||
name: str, arguments: dict | None
|
||||
) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
|
||||
"""
|
||||
Handle tool execution requests.
|
||||
Tools can modify server state and notify clients of changes.
|
||||
"""
|
||||
if name != "add-note":
|
||||
raise ValueError(f"Unknown tool: {name}")
|
||||
|
||||
if not arguments:
|
||||
raise ValueError("Missing arguments")
|
||||
|
||||
note_name = arguments.get("name")
|
||||
content = arguments.get("content")
|
||||
|
||||
if not note_name or not content:
|
||||
raise ValueError("Missing name or content")
|
||||
|
||||
# Update server state
|
||||
self.notes[note_name] = content
|
||||
|
||||
# Notify clients that resources have changed
|
||||
await self.request_context.session.send_resource_list_changed()
|
||||
|
||||
return [
|
||||
types.TextContent(
|
||||
return types.GetPromptResult(
|
||||
description="Summarize the current notes",
|
||||
messages=[
|
||||
types.PromptMessage(
|
||||
role="user",
|
||||
content=types.TextContent(
|
||||
type="text",
|
||||
text=f"Added note '{note_name}' with content: {content}",
|
||||
)
|
||||
]
|
||||
text=f"Here are the current notes to summarize:{detail_prompt}\n\n"
|
||||
+ "\n".join(
|
||||
f"- {name}: {content}"
|
||||
for name, content in notes.items()
|
||||
),
|
||||
),
|
||||
)
|
||||
],
|
||||
)
|
||||
|
||||
@server.list_tools()
|
||||
async def handle_list_tools() -> list[types.Tool]:
|
||||
"""
|
||||
List available tools.
|
||||
Each tool specifies its arguments using JSON Schema validation.
|
||||
"""
|
||||
return [
|
||||
types.Tool(
|
||||
name="add-note",
|
||||
description="Add a new note",
|
||||
inputSchema={
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"name": {"type": "string"},
|
||||
"content": {"type": "string"},
|
||||
},
|
||||
"required": ["name", "content"],
|
||||
},
|
||||
)
|
||||
]
|
||||
|
||||
@server.call_tool()
|
||||
async def handle_call_tool(
|
||||
name: str, arguments: dict | None
|
||||
) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
|
||||
"""
|
||||
Handle tool execution requests.
|
||||
Tools can modify server state and notify clients of changes.
|
||||
"""
|
||||
if name != "add-note":
|
||||
raise ValueError(f"Unknown tool: {name}")
|
||||
|
||||
if not arguments:
|
||||
raise ValueError("Missing arguments")
|
||||
|
||||
note_name = arguments.get("name")
|
||||
content = arguments.get("content")
|
||||
|
||||
if not note_name or not content:
|
||||
raise ValueError("Missing name or content")
|
||||
|
||||
# Update server state
|
||||
notes[note_name] = content
|
||||
|
||||
# Notify clients that resources have changed
|
||||
await server.request_context.session.send_resource_list_changed()
|
||||
|
||||
return [
|
||||
types.TextContent(
|
||||
type="text",
|
||||
text=f"Added note '{note_name}' with content: {content}",
|
||||
)
|
||||
]
|
||||
|
||||
async def main():
|
||||
server = McpServer()
|
||||
|
||||
# Run the server using stdin/stdout streams
|
||||
async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
|
||||
await server.run(
|
||||
|
|
|
|||
2
uv.lock
generated
2
uv.lock
generated
|
|
@ -24,7 +24,7 @@ wheels = [
|
|||
|
||||
[[package]]
|
||||
name = "create-mcp-server"
|
||||
version = "1.0.3.dev0"
|
||||
version = "1.0.4.dev0"
|
||||
source = { editable = "." }
|
||||
dependencies = [
|
||||
{ name = "click" },
|
||||
|
|
|
|||
Loading…
Reference in a new issue