diff --git a/src/create_mcp_server/__init__.py b/src/create_mcp_server/__init__.py index b54038d..df5bd5a 100644 --- a/src/create_mcp_server/__init__.py +++ b/src/create_mcp_server/__init__.py @@ -121,13 +121,16 @@ def get_package_directory(path: Path) -> Path: def copy_template( - path: Path, name: str, description: str, version: str = "0.1.0" + path: Path, name: str, description: str, version: str = "0.1.0", template_name: str = "blank" ) -> None: """Copy template files into src/""" template_dir = Path(__file__).parent / "template" + if template_name == "notes": + template_dir = template_dir / "notes" target_dir = get_package_directory(path) + import jinja2 from jinja2 import Environment, FileSystemLoader env = Environment(loader=FileSystemLoader(str(template_dir))) @@ -151,7 +154,7 @@ def copy_template( try: for template_file, output_file, output_dir in files: - template = env.get_template(template_file) + template: jinja2.Template = env.get_template(template_file) rendered = template.render(**template_vars) out_path = output_dir / output_file @@ -277,6 +280,12 @@ def check_package_name(name: str) -> bool: type=str, help="Project description", ) +@click.option( + "--template", + type=click.Choice(["blank", "notes"]), + default="blank", + help="Project template to use", +) @click.option( "--claudeapp/--no-claudeapp", default=True, @@ -287,6 +296,7 @@ def main( name: str | None, version: str | None, description: str | None, + template: str, claudeapp: bool, ) -> int: """Create a new MCP server project""" @@ -343,6 +353,7 @@ def main( project_path = project_path.resolve() create_project(project_path, name, description, version, claudeapp) + copy_template(project_path, name, description, version, template) update_pyproject_settings(project_path, version, description) return 0 diff --git a/src/create_mcp_server/template/notes/server.py.jinja2 b/src/create_mcp_server/template/notes/server.py.jinja2 new file mode 100644 index 0000000..33022b8 --- /dev/null +++ b/src/create_mcp_server/template/notes/server.py.jinja2 @@ -0,0 +1,164 @@ +import asyncio + +from mcp.server.models import InitializationOptions +import mcp.types as types +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] = {} + +server = Server("{{server_name}}") + +@server.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://internal/{name}"), + name=f"Note: {name}", + description=f"A simple note named {name}", + mimeType="text/plain", + ) + for name in notes + ] + +@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}") + + 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() + ), + ), + ) + ], + ) + +@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(): + # Run the server using stdin/stdout streams + async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): + await server.run( + read_stream, + write_stream, + InitializationOptions( + server_name="{{server_name}}", + server_version="{{server_version}}", + capabilities=server.get_capabilities( + notification_options=NotificationOptions(), + experimental_capabilities={}, + ), + ), + ) \ No newline at end of file diff --git a/src/create_mcp_server/template/server.py.jinja2 b/src/create_mcp_server/template/server.py.jinja2 index bc39f5c..ffc2991 100644 --- a/src/create_mcp_server/template/server.py.jinja2 +++ b/src/create_mcp_server/template/server.py.jinja2 @@ -3,149 +3,68 @@ import asyncio from mcp.server.models import InitializationOptions import mcp.types as types 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] = {} - server = Server("{{server_name}}") @server.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 [ + """List available resources. + + Example return value: + [ types.Resource( - uri=AnyUrl(f"note://internal/{name}"), - name=f"Note: {name}", - description=f"A simple note named {name}", - mimeType="text/plain", + uri=AnyUrl("resource://example"), + name="Example Resource", + description="An example resource", + mimeType="text/plain" ) - for name in notes ] - -@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}") - - name = uri.path - if name is not None: - name = name.lstrip("/") - return notes[name] - raise ValueError(f"Note not found: {name}") + return [] @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 [ + """List available prompts. + + Example return value: + [ types.Prompt( - name="summarize-notes", - description="Creates a summary of all notes", + name="example-prompt", + description="An example prompt", arguments=[ types.PromptArgument( name="style", - description="Style of the summary (brief/detailed)", - required=False, + description="Style of the prompt", + 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() - ), - ), - ) - ], - ) + return [] @server.list_tools() async def handle_list_tools() -> list[types.Tool]: - """ - List available tools. - Each tool specifies its arguments using JSON Schema validation. - """ - return [ + """List available tools. + + Example return value: + [ types.Tool( - name="add-note", - description="Add a new note", + name="example-tool", + description="An example tool", inputSchema={ "type": "object", "properties": { "name": {"type": "string"}, - "content": {"type": "string"}, + "value": {"type": "string"} }, - "required": ["name", "content"], - }, + "required": ["name", "value"] + } ) ] - -@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}", - ) - ] + return [] async def main(): # Run the server using stdin/stdout streams @@ -161,4 +80,4 @@ async def main(): experimental_capabilities={}, ), ), - ) + ) \ No newline at end of file