diff --git a/src/create_mcp_server/template/server.py.jinja2 b/src/create_mcp_server/template/server.py.jinja2 index bd07d11..bc39f5c 100644 --- a/src/create_mcp_server/template/server.py.jinja2 +++ b/src/create_mcp_server/template/server.py.jinja2 @@ -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}}") - - # Store notes as a simple key-value dict to demonstrate state management - self.notes: dict[str, str] = {} - - # 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", - ) - 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}") - - name = uri.path - if name is not None: - name = name.lstrip("/") - return self.notes[name] - raise ValueError(f"Note not found: {name}") - - # 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() - ), - ), - ) - ], - ) +@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}") - # 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"], - }, + name = uri.path + if name is not None: + name = name.lstrip("/") + return notes[name] + raise ValueError(f"Note not found: {name}") + +@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, ) - ] + ], + ) + ] + +@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}") + + 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 notes.items() + ), + ), + ) + ], + ) - @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}") +@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"], + }, + ) + ] - if not arguments: - raise ValueError("Missing arguments") +@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}") - note_name = arguments.get("name") - content = arguments.get("content") + if not arguments: + raise ValueError("Missing arguments") - if not note_name or not content: - raise ValueError("Missing name or content") + note_name = arguments.get("name") + content = arguments.get("content") - # Update server state - self.notes[note_name] = content + if not note_name or not content: + raise ValueError("Missing name or content") - # Notify clients that resources have changed - await self.request_context.session.send_resource_list_changed() + # Update server state + notes[note_name] = content - return [ - types.TextContent( - type="text", - text=f"Added note '{note_name}' with content: {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( diff --git a/uv.lock b/uv.lock index 372f52f..5e0d240 100644 --- a/uv.lock +++ b/uv.lock @@ -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" },