enhanced-mcp-tools/examples/component_service_examples.py
Ryan Malloy 8ff3775562 🚀 Major FastMCP 2.12.3 upgrade with ComponentService and BulkToolCaller integration
This massive update transforms Enhanced MCP Tools into a comprehensive workflow orchestration platform:

**Core Upgrades:**
- Updated FastMCP from 2.8.1 to 2.12.3 (latest release)
- Updated MCP SDK from 1.9.4 to 1.14.1
- Updated 29+ dependencies for compatibility

**New Features:**
- ComponentService integration with progressive tool disclosure
- SecurityManager with SACRED TRUST safety framework enhancement
- BulkToolCaller for workflow orchestration and batch operations
- Enhanced CLI with stdio default and explicit HTTP mode (--http flag)

**Security Enhancements:**
- Progressive tool disclosure (SAFE/CAUTION/DESTRUCTIVE levels)
- Safe mode enabled by default
- Destructive tools require explicit confirmation
- Mandatory dry-run validation for bulk operations
- Centralized security management across all modules

**Architecture Improvements:**
- Enhanced MCPBase with ComponentService integration
- Tool executor registry for bulk operations
- Backward compatibility with legacy modules
- Graceful fallback for missing ComponentService features

**Tool Count Expansion:**
- Total tools: 64+ (up from 50+)
- Categories: 16 (up from 14)
- New SecurityManager: 5 tools
- New BulkOperations: 8 tools

**Files Added:**
- src/enhanced_mcp/security_manager.py - Comprehensive security management
- src/enhanced_mcp/bulk_operations.py - Workflow orchestration system
- examples/ - Comprehensive integration guides and examples

**Files Modified:**
- pyproject.toml - FastMCP 2.12.3 dependency update
- src/enhanced_mcp/mcp_server.py - ComponentService integration
- src/enhanced_mcp/base.py - Enhanced MCPBase with security framework
- Multiple modules updated for ComponentService compatibility

All features tested and verified working. Server maintains stdio default behavior
for MCP clients while providing powerful workflow orchestration capabilities.
2025-09-22 17:15:02 -06:00

239 lines
8.4 KiB
Python

"""
ComponentService Integration Examples
This file demonstrates how to use the enhanced Enhanced MCP Tools server
with ComponentService integration for progressive tool disclosure and
security-focused dynamic tool management.
"""
import asyncio
from enhanced_mcp.mcp_server import create_server
from enhanced_mcp.base import SecurityLevel, ToolCategory
async def demo_progressive_tool_disclosure():
"""
Demonstrates progressive tool disclosure and security management
"""
print("🚀 Enhanced MCP Tools - ComponentService Integration Demo")
print("=" * 60)
# Create server with ComponentService integration
app = create_server("Demo Enhanced MCP Server")
print("\n🔍 Demo: Progressive Tool Disclosure")
print("-" * 40)
# Get initial tool list (should show only safe tools by default)
tools = await app.get_tools()
print(f"📋 Initial tools available: {len(tools)}")
# Show security manager tools
security_tools = [tool for tool in tools if tool.startswith("security_manager")]
print(f"🛡️ Security management tools: {len(security_tools)}")
for tool in security_tools:
print(f"{tool}")
# Show file operation tools that are visible
file_tools = [tool for tool in tools if tool.startswith("file_ops")]
print(f"📁 File operation tools: {len(file_tools)}")
for tool in file_tools:
print(f"{tool}")
print("\n✅ ComponentService integration successful!")
print("🛡️ Safe mode active - only safe tools visible by default")
async def demo_security_tool_categories():
"""
Demonstrates tool categorization by security level
"""
print("\n🏷️ Demo: Security Tool Categories")
print("-" * 40)
# Example tool categorizations
categories = {
SecurityLevel.SAFE: [
"watch_files", "list_archive_contents", "get_tool_info",
"security_status", "list_tools_by_security"
],
SecurityLevel.CAUTION: [
"file_backup", "create_archive", "git_commit",
"enable_destructive_tools"
],
SecurityLevel.DESTRUCTIVE: [
"bulk_rename", "extract_archive", "search_and_replace_batch",
"bulk_file_operations"
]
}
for level, tools in categories.items():
print(f"\n{level.upper()} Tools ({len(tools)}):")
for tool in tools:
print(f"{tool}")
print(f"\n📊 Total categorized tools: {sum(len(tools) for tools in categories.values())}")
def demo_tag_based_filtering():
"""
Demonstrates tag-based tool filtering capabilities
"""
print("\n🏷️ Demo: Tag-Based Tool Filtering")
print("-" * 40)
# Example tag filters
tag_filters = {
"readonly": ["watch_files", "list_archive_contents", "security_status"],
"destructive": ["bulk_rename", "extract_archive", "search_and_replace_batch"],
"filesystem": ["bulk_rename", "file_backup", "watch_files"],
"archive": ["create_archive", "extract_archive", "list_archive_contents"],
"security": ["enable_destructive_tools", "set_safe_mode", "security_status"],
"bulk_operations": ["bulk_rename", "search_and_replace_batch"]
}
for tag, tools in tag_filters.items():
print(f"\n'{tag}' tagged tools ({len(tools)}):")
for tool in tools:
print(f"{tool}")
async def demo_dynamic_tool_control():
"""
Demonstrates dynamic tool visibility control
"""
print("\n🎛️ Demo: Dynamic Tool Control")
print("-" * 40)
app = create_server("Dynamic Control Demo")
print("1. Initial state: Safe mode enabled")
initial_tools = await app.get_tools()
safe_tools = [t for t in initial_tools if not any(
keyword in t.lower() for keyword in ['bulk', 'destructive', 'delete']
)]
print(f" Safe tools visible: {len(safe_tools)}")
print("\n2. Simulated: Enable destructive tools")
print(" Command: security_manager_enable_destructive_tools(enabled=True, confirm_destructive=True)")
print(" Result: Destructive tools become visible")
print("\n3. Simulated: Disable safe mode")
print(" Command: security_manager_set_safe_mode(safe_mode=False)")
print(" Result: All non-destructive tools become visible")
print("\n4. Progressive disclosure flow:")
print(" a) Start: Only safe tools (monitoring, read-only)")
print(" b) Normal: Add caution tools (create/modify, reversible)")
print(" c) Advanced: Add destructive tools (bulk ops, irreversible)")
def demo_best_practices():
"""
Demonstrates best practices for using ComponentService integration
"""
print("\n🎯 Demo: Best Practices")
print("-" * 40)
practices = [
{
"title": "1. Security-First Design",
"description": "Start with safe tools, progressively enable more powerful ones",
"example": "Always begin with security_manager tools to assess available capabilities"
},
{
"title": "2. Explicit Confirmation",
"description": "Require confirmation for destructive operations",
"example": "enable_destructive_tools(enabled=True, confirm_destructive=True)"
},
{
"title": "3. Dry Run First",
"description": "Always test destructive operations with dry_run=True",
"example": "bulk_rename(pattern='old', replacement='new', dry_run=True)"
},
{
"title": "4. Tool Discovery",
"description": "Use security manager to explore available tools safely",
"example": "list_tools_by_security() to see what's available at each level"
},
{
"title": "5. Category-Based Access",
"description": "Enable tools by category as needed",
"example": "Show only file operations, or only archive tools"
}
]
for practice in practices:
print(f"\n{practice['title']}:")
print(f" {practice['description']}")
print(f" Example: {practice['example']}")
async def demo_integration_patterns():
"""
Demonstrates integration patterns with existing MCPMixin modules
"""
print("\n🔧 Demo: Integration Patterns")
print("-" * 40)
patterns = [
{
"pattern": "Multiple Inheritance",
"description": "class MyModule(MCPMixin, MCPBase)",
"benefit": "Combines FastMCP tools with security features"
},
{
"pattern": "Tool Metadata Registration",
"description": "self.register_tagged_tool(name, security_level, category, tags)",
"benefit": "Enables progressive disclosure and categorization"
},
{
"pattern": "Enhanced Registration",
"description": "module.safe_register_all(app, prefix='module')",
"benefit": "Automatic ComponentService setup and security filtering"
},
{
"pattern": "Centralized Security Control",
"description": "security_manager.register_tool_module(name, module)",
"benefit": "Single point of control for all tool visibility"
},
{
"pattern": "Backward Compatibility",
"description": "Fallback to legacy registration if enhanced fails",
"benefit": "Smooth migration path for existing tools"
}
]
for pattern in patterns:
print(f"\n{pattern['pattern']}:")
print(f" Implementation: {pattern['description']}")
print(f" Benefit: {pattern['benefit']}")
async def main():
"""
Run all ComponentService integration demos
"""
try:
await demo_progressive_tool_disclosure()
await demo_security_tool_categories()
demo_tag_based_filtering()
await demo_dynamic_tool_control()
demo_best_practices()
await demo_integration_patterns()
print("\n" + "=" * 60)
print("✅ All ComponentService demos completed successfully!")
print("\n📖 Next Steps:")
print("1. Run: uvx enhanced-mcp-tools --list-tools")
print("2. Start server: uvx enhanced-mcp-tools")
print("3. Use security_manager tools to control tool visibility")
print("4. Enable destructive tools only when needed with confirmation")
except Exception as e:
print(f"❌ Demo error: {e}")
print("💡 Ensure FastMCP and enhanced-mcp-tools are properly installed")
if __name__ == "__main__":
asyncio.run(main())