MCP转OpenAPI代理增强版

from mcpo_enhanced import ConfigAnalyzer, SecurityAnalyzer, ToolSchemaAnalyzer

config = {
    "mcpServers": {
        "filesystem": {
            "command": "npx",
            "args": ["-y", "@modelcontextprotocol/server-filesystem", "/tmp"]
        },
        "web": {
            "type": "sse",
            "url": "http://localhost:8080/sse"
        }
    }
}

# Analyze configuration structure
analyzer = ConfigAnalyzer(config)
print(analyzer.server_count())        # 2
print(analyzer.transport_types())     # {'stdio': 1, 'sse': 1}
print(analyzer.list_server_names())   # ['filesystem', 'web']

# Security audit
security = SecurityAnalyzer(config)
report = security.full_audit()
print(report["risk_score"])           # 0-10 risk rating
print(report["api_key_analysis"])     # Key strength assessment

# Tool schema analysis
tools = [{"name": "read_file", "inputSchema": {"type": "object", "properties": {"path": {"type": "string"}}}}]
schema_analyzer = ToolSchemaAnalyzer(tools)
print(schema_analyzer.complexity_scores())     # Per-tool complexity 0-10
print(schema_analyzer.detect_dangerous_tools()) # Flags risky operations

from mcpo_enhanced import ToolFilter, ServerRouter, ConfigGenerator

tools = [
    {"name": "read_file", "inputSchema": {"type": "object", "properties": {"path": {"type": "string"}}}},
    {"name": "execute_sql", "inputSchema": {"type": "object", "properties": {"query": {"type": "string"}, "db": {"type": "string"}}}},
    {"name": "list_dir", "inputSchema": {"type": "object", "properties": {}}},
]

# Filter tools by criteria
tf = ToolFilter(tools)
safe_tools = tf.exclude_dangerous()          # Remove risky tools
simple = tf.filter_by_max_params(1)          # Tools with <= 1 param
matches = tf.filter_by_name_pattern("file")  # Name-based filtering

# Route requests to appropriate servers
config = {"mcpServers": {"fs": {"command": "fs-server"}, "db": {"command": "db-server"}}}
router = ServerRouter(config)
server = router.route_by_capability(["read_file"])  # -> "fs"

# Generate optimized configurations
gen = ConfigGenerator(config)
minimal = gen.generate_minimal()         # Strip optional fields
secured = gen.with_api_key("my-key")     # Add authentication

from mcpo_enhanced import SchemaComplexityMetrics, TransportBenchmark, CapacityEstimator

tools = [
    {"name": "simple_tool", "inputSchema": {"type": "object", "properties": {"x": {"type": "string"}}}},
    {"name": "complex_tool", "inputSchema": {"type": "object", "properties": {
        "query": {"type": "string"}, "filters": {"type": "object"},
        "options": {"type": "array", "items": {"type": "string"}}
    }}}
]

# Measure schema complexity
metrics = SchemaComplexityMetrics(tools)
scores = metrics.complexity_scores()          # Per-tool 0-10
print(metrics.histogram(bins=5))              # Distribution
print(metrics.rank_by_complexity()[:3])        # Top 3 complex tools

# Compare transport performance
bench = TransportBenchmark()
comparison = bench.compare_transports()       # stdio vs sse vs http
print(bench.recommend_transport(tool_count=50, concurrent_users=10))

# Estimate capacity
config = {"mcpServers": {"a": {"command": "a"}, "b": {"type": "sse", "url": "http://x"}}}
estimator = CapacityEstimator(config, tools)
print(estimator.estimate_memory_mb())         # Memory estimate
print(estimator.estimate_endpoints())         # Endpoint count
print(estimator.scaling_recommendations(target_rps=100))

git clone https://github.com/Rlin1027/mcpo-enhanced.git
cd mcpo-enhanced
pip install -e .     # Install original mcpo

python3 -m pytest tests/ -v

mcpo-enhanced/
  src/mcpo/              # Original mcpo source (untouched)
  mcpo_enhanced/         # New enhancement package
    __init__.py          # 15 public classes
    analysis.py          # Configuration & security analysis
    routing.py           # Tool filtering & server routing
    benchmark.py         # Performance benchmarking
  tests/
    test_analysis.py     # 44 tests
    test_routing.py      # 52 tests
    test_benchmark.py    # 63 tests