template.py.j2•7.58 kB
"""
{{ server_name.replace('_', ' ').title() }} - {{ description }}
A clean, well-structured MCP server demonstrating best practices.
Generated by MCP-Creator-MCP
"""
import asyncio
import atexit
import logging
import atexit
import logging
import signal
import sys
from typing import Any
from mcp.server.fastmcp import FastMCP, Context
# Configure logging to stderr for MCP compliance
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
stream=sys.stderr,
)
logger = logging.getLogger(__name__)
# Initialize MCP server with clean interface
mcp = FastMCP("{{ server_name }}")
# Global state management
_server_state: dict[str, object] = {}
{% if 'tools' in features %}
@mcp.tool()
async def example_tool(ctx: Context, input_text: str) -> str:
"""
Process input text with intelligent handling.
Demonstrates clean tool implementation with proper error handling,
validation, and user-friendly responses.
Args:
input_text: Text to process
Returns:
Processed result with clear status indicators
"""
try:
# Input validation with clear feedback
if not input_text or not input_text.strip():
return "⚠️ Please provide valid input text"
# Core processing logic
processed_text = input_text.strip().upper()
result = f"✅ Processed: {processed_text}"
# Logging for operational visibility
logger.info(f"Successfully processed input: {len(input_text)} characters")
return result
except Exception as e:
# Graceful error handling with helpful messages
logger.error(f"Tool execution failed: {e}")
return f"❌ Processing failed: {str(e)}"
@mcp.tool()
async def get_server_status(ctx: Context) -> str:
"""
Retrieve current server status and health information.
Provides operational visibility and demonstrates resource
access patterns with clean state management.
Returns:
Current server status and metrics
"""
try:
status_info = {
"server": "{{ server_name }}",
"status": "healthy",
"features": {{ features | tojson }},
"uptime": "operational"
}
# Format for human readability
status_text = "\n".join(f"• {k}: {v}" for k, v in status_info.items())
return f"📊 Server Status:\n{status_text}"
except Exception as e:
logger.error(f"Status check failed: {e}")
return f"❌ Status unavailable: {str(e)}"
{% endif %}
{% if 'resources' in features %}
@mcp.resource("resource://{{ server_name }}/info")
async def server_info() -> str:
"""
Provide server information and capabilities.
Demonstrates static resource pattern with structured data
that enhances LLM context understanding.
"""
try:
info = {
"name": "{{ server_name }}",
"description": "{{ description }}",
"version": "1.0.0",
"capabilities": {{ features | tojson }},
"generated_by": "MCP-Creator-MCP"
}
# Return structured data for LLM consumption
import json
return json.dumps(info, indent=2)
except Exception as e:
logger.error(f"Resource access failed: {e}")
return '{"error": "Resource unavailable: " + str(e) + "}"}'
@mcp.resource("resource://{{ server_name }}/help/{topic}")
async def help_resource(topic: str) -> str:
"""
Provide contextual help for different topics.
Demonstrates dynamic resource pattern with parameter validation
and intelligent fallback behavior.
Args:
topic: Help topic to retrieve
Returns:
Help content for the specified topic
"""
try:
help_content = {
"usage": "This server provides example tools and resources for learning MCP development.",
"tools": "Use example_tool to process text and get_server_status for health checks.",
"resources": "Access server_info for capabilities and help/{topic} for guidance.",
"support": "Check the server logs for detailed operational information."
}
content = help_content.get(
topic.lower(),
f"Help topic '{topic}' not found. Available topics: {', '.join(help_content.keys())}"
)
return content
except Exception as e:
logger.error(f"Help resource failed: {e}")
return f"Help unavailable: {str(e)}"
{% endif %}
{% if 'prompts' in features %}
@mcp.prompt()
def analysis_prompt(subject: str, focus_area: str = "general") -> str:
"""
Generate analysis prompt template with intelligent customization.
Demonstrates clean prompt pattern with parameter validation
and contextual content generation.
Args:
subject: Subject to analyze
focus_area: Specific area of focus for analysis
Returns:
Formatted analysis prompt
"""
try:
# Validate inputs with helpful feedback
if not subject:
return "Please provide a subject for analysis."
# Contextual prompt generation
prompt_template = f"""
Please provide a comprehensive analysis of: {subject}
Focus Area: {focus_area}
Analysis Framework:
1. Overview and context
2. Key characteristics and patterns
3. Strengths and opportunities
4. Potential challenges or limitations
5. Recommendations and next steps
Please provide specific, actionable insights with supporting evidence.
"""
return prompt_template.strip()
except Exception as e:
logger.error(f"Prompt generation failed: {e}")
return f"Prompt generation error: {str(e)}"
{% endif %}
# Resource cleanup and lifecycle management
def cleanup_resources() -> None:
"""
Clean up server resources on shutdown.
Demonstrates proper resource management patterns that prevent
process leaks and ensure graceful server termination.
"""
logger.info("Cleaning up {{ server_name }} server resources")
# Clear global state
_server_state.clear()
# Cancel any background tasks
try:
# Example: cancel background tasks, close connections
logger.info("Resource cleanup completed")
except Exception as e:
logger.error(f"Cleanup error: {e}")
def signal_handler(signum: int, frame) -> None:
"""
Handle shutdown signals gracefully.
Ensures proper cleanup and logging for operational visibility.
"""
logger.info(f"Received signal {signum}, initiating graceful shutdown")
cleanup_resources()
sys.exit(0)
# Register signal handlers for graceful shutdown
signal.signal(signal.SIGTERM, signal_handler)
signal.signal(signal.SIGINT, signal_handler)
atexit.register(cleanup_resources)
if __name__ == "__main__":
try:
logger.info("Starting {{ server_name }} MCP server")
logger.info(f"Features enabled: {{ features | join(', ') }}")
# Start server with proper error handling
mcp.run()
except KeyboardInterrupt:
logger.info("Server interrupted by user")
except Exception as e:
logger.error(f"Server startup failed: {e}")
sys.exit(1)
finally:
cleanup_resources()