Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat: add template selection support #11

Open
wants to merge 10 commits into
base: main
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 13 additions & 2 deletions src/create_mcp_server/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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/<project_name>"""
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)))
Expand All @@ -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
Expand Down Expand Up @@ -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,
Expand All @@ -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"""
Expand Down Expand Up @@ -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
164 changes: 164 additions & 0 deletions src/create_mcp_server/template/notes/server.py.jinja2
Original file line number Diff line number Diff line change
@@ -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={},
),
),
)
Loading
Loading