rsp2k
ddf23d10e4
compose_feedback_services() called import_server() without awaiting it. That method became async in FastMCP 2.14, so on the version this package was built against it composed nothing while logging success. FastMCP 4 removed it outright. Use mount(), which is sync on 2.12 through 4.x. The test suite could not start (pytest-html hook, plugin not installed), so nothing caught this. Add pytest-html, bring examples/ over from the docs repo since the integration tests import them, and fix the tests that had drifted from the code. 407 tests pass on FastMCP 2.12.2, 2.14.3 and 4.0.10. - Declare fastmcp>=2.12.2,<5; CI matrix covers each family plus a non-blocking canary on the newest FastMCP - Add a composition test with real servers; the mocked ones could not see a missing await - Rewrite the README against the actual API and execute its examples - Ignore ruff UP042: StrEnum would change stored enum formatting - Version 2026.09.27
403 lines
14 KiB
Python
403 lines
14 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
FastMCP Feedback Demo Server
|
||
|
||
A comprehensive demonstration server that showcases all features of the
|
||
FastMCP Feedback system including:
|
||
- Complete feedback collection workflow
|
||
- Privacy-compliant analytics
|
||
- Multi-database support
|
||
- Performance monitoring
|
||
- Server composition patterns
|
||
|
||
Usage:
|
||
# Run the demo server
|
||
python examples/demo_server.py
|
||
|
||
# Or use uv
|
||
uv run examples/demo_server.py
|
||
|
||
# Connect from Claude Desktop with this MCP server configuration:
|
||
# {
|
||
# "feedback-demo": {
|
||
# "command": "uv",
|
||
# "args": ["run", "examples/demo_server.py"]
|
||
# }
|
||
# }
|
||
"""
|
||
|
||
import asyncio
|
||
import logging
|
||
import sys
|
||
import tempfile
|
||
from datetime import UTC, datetime
|
||
from pathlib import Path
|
||
|
||
# Add the src directory to the path for development
|
||
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
||
|
||
from fastmcp import Client, FastMCP
|
||
|
||
from fastmcp_feedback import (
|
||
FeedbackDatabase,
|
||
FeedbackInsights,
|
||
add_feedback_tools,
|
||
check_dependencies,
|
||
get_version_info,
|
||
)
|
||
from fastmcp_feedback.feedback import (
|
||
create_analytics_server,
|
||
create_management_server,
|
||
create_submission_server,
|
||
)
|
||
|
||
# Configure logging
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
|
||
handlers=[logging.StreamHandler(), logging.FileHandler("feedback_demo.log")],
|
||
)
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
def create_temp_database() -> str:
|
||
"""Create a temporary SQLite database for the demo."""
|
||
temp_dir = tempfile.mkdtemp(prefix="fastmcp_feedback_demo_")
|
||
db_path = Path(temp_dir) / "demo_feedback.db"
|
||
return f"sqlite:///{db_path}"
|
||
|
||
|
||
async def setup_demo_data(database: FeedbackDatabase) -> None:
|
||
"""Populate the database with sample feedback for demonstration."""
|
||
logger.info("Setting up demo data...")
|
||
|
||
sample_feedback = [
|
||
{
|
||
"type": "bug",
|
||
"title": "Memory leak in analytics processing",
|
||
"description": "Analytics worker consumes increasing memory over time, eventually causing OOM errors.",
|
||
"submitter": "alice@example.com",
|
||
"contact_info": "alice@engineering.example.com",
|
||
},
|
||
{
|
||
"type": "feature",
|
||
"title": "Add export functionality for feedback data",
|
||
"description": "Users need ability to export feedback data as CSV or JSON for external analysis.",
|
||
"submitter": "product_manager",
|
||
"contact_info": "pm@example.com",
|
||
},
|
||
{
|
||
"type": "improvement",
|
||
"title": "Optimize database query performance",
|
||
"description": "List feedback queries are slow with large datasets. Need pagination and indexing improvements.",
|
||
"submitter": "performance_team",
|
||
},
|
||
{
|
||
"type": "question",
|
||
"title": "Best practices for rate limiting?",
|
||
"description": "What are recommended rate limits for feedback submission to prevent abuse?",
|
||
"submitter": "security_team",
|
||
"contact_info": "security@example.com",
|
||
},
|
||
{
|
||
"type": "bug",
|
||
"title": "Unicode handling in feedback titles",
|
||
"description": "Emojis and special characters cause encoding errors in feedback titles. 🐛",
|
||
"submitter": "qa_engineer",
|
||
},
|
||
]
|
||
|
||
try:
|
||
await database.initialize()
|
||
|
||
# Import the model here to avoid circular imports
|
||
from fastmcp_feedback.models import Feedback, FeedbackStatus, FeedbackType
|
||
|
||
async with database.session() as session:
|
||
for item in sample_feedback:
|
||
feedback = Feedback(
|
||
type=FeedbackType(item["type"]),
|
||
status=FeedbackStatus.OPEN,
|
||
title=item["title"],
|
||
description=item["description"],
|
||
submitter=item["submitter"],
|
||
contact_info=item.get("contact_info"),
|
||
created_at=datetime.now(UTC),
|
||
updated_at=datetime.now(UTC),
|
||
)
|
||
session.add(feedback)
|
||
|
||
session.commit()
|
||
logger.info(f"Added {len(sample_feedback)} sample feedback items")
|
||
|
||
except Exception as e:
|
||
logger.error(f"Failed to setup demo data: {e}")
|
||
raise
|
||
|
||
|
||
def create_demo_server(database_url: str | None = None) -> FastMCP:
|
||
"""Create the main demo server with all feedback tools.
|
||
|
||
Uses a fresh temporary SQLite database unless database_url is given.
|
||
"""
|
||
logger.info("🚀 Creating FastMCP Feedback Demo Server")
|
||
|
||
# Create the main server
|
||
server = FastMCP("FastMCP Feedback Demo")
|
||
server.description = (
|
||
"Comprehensive demonstration of FastMCP Feedback system featuring "
|
||
"complete feedback lifecycle, analytics, and multi-database support."
|
||
)
|
||
|
||
# Create temporary database for demo
|
||
database_url = database_url or create_temp_database()
|
||
logger.info(f"Using demo database: {database_url}")
|
||
|
||
# Configure insights for demo (privacy-compliant by design)
|
||
insights = FeedbackInsights(
|
||
enabled=True,
|
||
retention_days=30, # Short retention for demo
|
||
)
|
||
|
||
# Add all feedback tools to the main server
|
||
add_feedback_tools(
|
||
server,
|
||
database_url=database_url,
|
||
insights=insights,
|
||
prefix="", # No prefix for main tools
|
||
)
|
||
|
||
# Add version info tool
|
||
@server.tool(description="Get FastMCP Feedback version and dependency information")
|
||
def get_system_info() -> dict:
|
||
"""Get comprehensive system information."""
|
||
version_info = get_version_info()
|
||
dependency_status = check_dependencies()
|
||
|
||
return {
|
||
"system": "FastMCP Feedback Demo Server",
|
||
"status": "running",
|
||
"started_at": datetime.now(UTC).isoformat(),
|
||
"versions": version_info,
|
||
"dependencies": dependency_status,
|
||
"demo_features": [
|
||
"Complete feedback lifecycle",
|
||
"Privacy-compliant analytics",
|
||
"Multi-database support",
|
||
"Performance monitoring",
|
||
"Sample data included",
|
||
],
|
||
}
|
||
|
||
# Add demo data setup tool
|
||
@server.tool(description="Populate database with sample feedback data")
|
||
async def setup_sample_data() -> dict:
|
||
"""Setup sample feedback data for demonstration."""
|
||
try:
|
||
database = FeedbackDatabase(database_url)
|
||
await setup_demo_data(database)
|
||
await database.close()
|
||
|
||
return {
|
||
"success": True,
|
||
"message": "Sample data created successfully",
|
||
"items_added": 5,
|
||
"data_includes": [
|
||
"Bug reports with technical details",
|
||
"Feature requests from product team",
|
||
"Performance improvement suggestions",
|
||
"Security questions",
|
||
"Unicode handling issues",
|
||
],
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"Failed to setup sample data: {e}")
|
||
return {"success": False, "error": str(e)}
|
||
|
||
return server
|
||
|
||
|
||
async def create_composed_demo_server() -> FastMCP:
|
||
"""Create a demo server using server composition patterns."""
|
||
logger.info("🏗️ Creating Composed Demo Server")
|
||
|
||
# Create main application server
|
||
main_server = FastMCP("Composed Feedback System")
|
||
main_server.description = "Demonstration of FastMCP server composition patterns"
|
||
|
||
# Create temporary databases for different services
|
||
public_db = create_temp_database()
|
||
admin_db = create_temp_database()
|
||
analytics_db = create_temp_database()
|
||
|
||
# Create specialized servers
|
||
public_server = create_submission_server(
|
||
"Public Feedback Portal",
|
||
database_url=public_db,
|
||
insights=FeedbackInsights(enabled=True, retention_days=90),
|
||
)
|
||
|
||
admin_server = create_management_server(
|
||
"Admin Management Panel",
|
||
database_url=admin_db,
|
||
insights=FeedbackInsights(enabled=True, retention_days=365),
|
||
)
|
||
|
||
analytics_server = create_analytics_server(
|
||
"Analytics Dashboard",
|
||
database_url=analytics_db,
|
||
insights=FeedbackInsights(enabled=True, retention_days=30),
|
||
)
|
||
|
||
# Compose servers with prefixes
|
||
main_server.mount(public_server, "public")
|
||
main_server.mount(admin_server, "admin")
|
||
main_server.mount(analytics_server, "analytics")
|
||
|
||
# Add composition info tool
|
||
@main_server.tool(description="Get information about server composition")
|
||
def get_composition_info() -> dict:
|
||
"""Get information about the composed server architecture."""
|
||
return {
|
||
"architecture": "Multi-server composition",
|
||
"servers": {
|
||
"public": {
|
||
"purpose": "Feedback submission for end users",
|
||
"tools": ["submit_feedback"],
|
||
"database": "isolated",
|
||
},
|
||
"admin": {
|
||
"purpose": "Management and moderation",
|
||
"tools": [
|
||
"update_feedback_status",
|
||
"delete_feedback",
|
||
"bulk_operations",
|
||
],
|
||
"database": "isolated",
|
||
},
|
||
"analytics": {
|
||
"purpose": "Reporting and insights",
|
||
"tools": ["generate_reports", "export_data", "usage_statistics"],
|
||
"database": "isolated",
|
||
},
|
||
},
|
||
"benefits": [
|
||
"Separation of concerns",
|
||
"Independent scaling",
|
||
"Role-based access control",
|
||
"Database isolation",
|
||
"Service modularity",
|
||
],
|
||
}
|
||
|
||
return main_server
|
||
|
||
|
||
async def run_demo_tests(server: FastMCP) -> dict:
|
||
"""Run basic functionality tests on the demo server."""
|
||
logger.info("🧪 Running demo functionality tests...")
|
||
|
||
try:
|
||
# Go through a client so this works the same on every FastMCP version
|
||
async with Client(server) as client:
|
||
tool_names = [tool.name for tool in await client.list_tools()]
|
||
|
||
system_info = None
|
||
if "get_system_info" in tool_names:
|
||
result = await client.call_tool("get_system_info", {})
|
||
system_info = result.structured_content
|
||
|
||
setup_result = None
|
||
if "setup_sample_data" in tool_names:
|
||
result = await client.call_tool("setup_sample_data", {})
|
||
setup_result = result.structured_content
|
||
|
||
return {
|
||
"test_status": "passed",
|
||
"tools_available": len(tool_names),
|
||
"tool_names": tool_names,
|
||
"system_info": system_info,
|
||
"sample_data_setup": setup_result,
|
||
"test_timestamp": datetime.now(UTC).isoformat(),
|
||
}
|
||
|
||
except Exception as e:
|
||
logger.error(f"Demo tests failed: {e}")
|
||
return {
|
||
"test_status": "failed",
|
||
"error": str(e),
|
||
"test_timestamp": datetime.now(UTC).isoformat(),
|
||
}
|
||
|
||
|
||
def print_demo_banner():
|
||
"""Print an informative banner for the demo server."""
|
||
banner = """
|
||
╔══════════════════════════════════════════════════════════════════╗
|
||
║ 🚀 FastMCP Feedback Demo Server ║
|
||
║ ║
|
||
║ Features Demonstrated: ║
|
||
║ • Complete feedback collection system ║
|
||
║ • Privacy-compliant analytics (GDPR-ready) ║
|
||
║ • Multi-database support (SQLite, PostgreSQL) ║
|
||
║ • Thread-safe operations with 97% test coverage ║
|
||
║ • Server composition patterns ║
|
||
║ • Performance monitoring and health checks ║
|
||
║ ║
|
||
║ Available Tools: ║
|
||
║ 📝 submit_feedback - Submit bug reports & feature requests║
|
||
║ 📋 list_feedback - Query and filter feedback ║
|
||
║ 📊 get_feedback_statistics - Analytics and metrics ║
|
||
║ 🔧 update_feedback_status - Change status (admin) ║
|
||
║ 🗑️ delete_feedback - Remove feedback items ║
|
||
║ ℹ️ get_system_info - Version and dependency info ║
|
||
║ 🎯 setup_sample_data - Populate with demo data ║
|
||
║ ║
|
||
║ Usage: Connect this as an MCP server in Claude Desktop ║
|
||
║ Configuration: See comments in this file ║
|
||
╚══════════════════════════════════════════════════════════════════╝
|
||
"""
|
||
print(banner)
|
||
|
||
|
||
async def main():
|
||
"""Main entry point for the demo server."""
|
||
print_demo_banner()
|
||
|
||
try:
|
||
# Check dependencies first
|
||
deps = check_dependencies()
|
||
if not deps["all_available"]:
|
||
logger.error("Missing required dependencies!")
|
||
for dep, info in deps["details"].items():
|
||
if not info["available"]:
|
||
logger.error(f" Missing: {dep} {info['required']}")
|
||
sys.exit(1)
|
||
|
||
logger.info("All dependencies satisfied ✅")
|
||
|
||
# Create the main demo server
|
||
server = create_demo_server()
|
||
|
||
# Run functionality tests
|
||
test_results = await run_demo_tests(server)
|
||
logger.info(f"Demo tests: {test_results['test_status']}")
|
||
|
||
if test_results["test_status"] == "passed":
|
||
logger.info(f"Server ready with {test_results['tools_available']} tools")
|
||
|
||
# Start the FastMCP server
|
||
logger.info("🎬 Starting FastMCP Feedback Demo Server...")
|
||
await server.run()
|
||
|
||
except KeyboardInterrupt:
|
||
logger.info("Demo server stopped by user")
|
||
except Exception as e:
|
||
logger.error(f"Demo server failed: {e}")
|
||
sys.exit(1)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main())
|