# ============================================================================= # root/app/mcp.py # Universal MCP Hub (Sandboxed) - based on PyFundaments Architecture # Copyright 2026 - Volkan Kücükbudak # Apache License V. 2 + ESOL 1.1 # Repo: https://github.com/VolkanSah/Universal-MCP-Hub-sandboxed # ============================================================================= # ARCHITECTURE NOTE: # This file lives exclusively in app/ and is ONLY started by app/app.py. # NO direct access to fundaments/*, .env, or Guardian (main.py). # All config comes from app/.pyfun via app/config.py. # # MCP SSE transport runs through Quart/hypercorn via /mcp route. # All MCP traffic can be intercepted, logged, and transformed in app.py # before reaching this handler — this is by design. # # TOOL REGISTRATION PRINCIPLE: # Tools are registered via tools.py — NOT hardcoded here. # No key = no provider = no tool = no crash. # Server always starts, just with fewer tools. # Adding a new tool = update .pyfun + providers.py only. Never touch mcp.py. # # DEPENDENCY CHAIN (app/* only, no fundaments!): # config.py → parses app/.pyfun — single source of truth # providers.py → LLM + Search provider registry + fallback chain # models.py → model limits, costs, capabilities from .pyfun [MODELS] # tools.py → tool registry + execution — reads .pyfun [TOOLS] # db_sync.py → internal SQLite IPC (app/* state) — NOT postgresql.py! # mcp.py → registers tools only, delegates all logic to tools.py # ============================================================================= import logging from typing import Dict, Any from . import config as app_config from . import providers from . import models from . import tools logger = logging.getLogger('mcp') # ============================================================================= # Global MCP instance — initialized once via initialize() # ============================================================================= _mcp = None # ============================================================================= # Initialization — called exclusively by app/app.py # ============================================================================= async def initialize() -> None: """ Initializes the MCP instance and registers all tools. Called once by app/app.py during startup sequence. No fundaments passed in — fully sandboxed. Registration order: 1. LLM tools → via tools.py + providers.py (key-gated) 2. Search tools → via tools.py + providers.py (key-gated) 3. System tools → always registered, no key required 4. DB tools → uncomment when db_sync.py is ready """ global _mcp logger.info("MCP Hub initializing...") hub_cfg = app_config.get_hub() try: from mcp.server.fastmcp import FastMCP except ImportError: logger.critical("FastMCP not installed. Run: pip install mcp") raise _mcp = FastMCP( name=hub_cfg.get("HUB_NAME", "Universal MCP Hub"), instructions=( f"{hub_cfg.get('HUB_DESCRIPTION', 'Universal MCP Hub on PyFundaments')} " "Use list_active_tools to see what is currently available." ) ) # --- Initialize registries --- providers.initialize() models.initialize() tools.initialize() # --- Register MCP tools --- _register_llm_tools(_mcp) _register_search_tools(_mcp) _register_system_tools(_mcp) # _register_db_tools(_mcp) # uncomment when db_sync.py is ready logger.info("MCP Hub initialized.") # ============================================================================= # Request Handler — Quart /mcp route entry point # ============================================================================= async def handle_request(request) -> None: """ Handles incoming MCP SSE requests routed through Quart /mcp endpoint. Central interceptor point for all MCP traffic. Add auth, logging, rate limiting, payload transformation here as needed. """ if _mcp is None: logger.error("MCP not initialized — call initialize() first.") from quart import jsonify return jsonify({"error": "MCP not initialized"}), 503 # --- Interceptor hooks (uncomment as needed) --- # logger.debug(f"MCP request: {request.method} {request.path}") # await _check_auth(request) # await _rate_limit(request) # await _log_payload(request) return await _mcp.handle_sse(request) # ============================================================================= # Tool Registration — delegates all logic to tools.py # ============================================================================= def _register_llm_tools(mcp) -> None: """ Register LLM completion tool. All logic delegated to tools.py → providers.py. Adding a new LLM provider = update .pyfun + providers.py. Never touch this. """ if not providers.list_active_llm(): logger.info("No active LLM providers — llm_complete tool skipped.") return @mcp.tool() async def llm_complete( prompt: str, provider: str = None, model: str = None, max_tokens: int = 1024, ) -> str: """ Send a prompt to any configured LLM provider. Automatically follows the fallback chain defined in .pyfun if a provider fails. Args: prompt: The input text to send to the model. provider: Provider name (e.g. 'anthropic', 'gemini', 'openrouter', 'huggingface'). Defaults to default_provider from .pyfun [TOOL.llm_complete]. model: Model name override. Defaults to provider's default_model in .pyfun. max_tokens: Maximum tokens in the response. Default: 1024. Returns: Model response as plain text string. """ return await tools.run( tool_name="llm_complete", prompt=prompt, provider_name=provider, model=model, max_tokens=max_tokens, ) logger.info(f"Tool registered: llm_complete (active providers: {providers.list_active_llm()})") def _register_search_tools(mcp) -> None: """ Register web search tool. All logic delegated to tools.py → providers.py. Adding a new search provider = update .pyfun + providers.py. Never touch this. """ if not providers.list_active_search(): logger.info("No active search providers — web_search tool skipped.") return @mcp.tool() async def web_search( query: str, provider: str = None, max_results: int = 5, ) -> str: """ Search the web via any configured search provider. Automatically follows the fallback chain defined in .pyfun if a provider fails. Args: query: Search query string. provider: Provider name (e.g. 'brave', 'tavily'). Defaults to default_provider from .pyfun [TOOL.web_search]. max_results: Maximum number of results to return. Default: 5. Returns: Formatted search results as plain text string. """ return await tools.run( tool_name="web_search", prompt=query, provider_name=provider, max_results=max_results, ) logger.info(f"Tool registered: web_search (active providers: {providers.list_active_search()})") def _register_system_tools(mcp) -> None: """ System tools — always registered, no ENV key required. Exposes hub status and model info without touching secrets. """ @mcp.tool() def list_active_tools() -> Dict[str, Any]: """ List all active providers and registered tools. Shows ENV key names only — never exposes values or secrets. Returns: Dict with hub info, active LLM providers, active search providers, available tools and model names. """ hub = app_config.get_hub() return { "hub": hub.get("HUB_NAME", "Universal MCP Hub"), "version": hub.get("HUB_VERSION", ""), "active_llm_providers": providers.list_active_llm(), "active_search_providers": providers.list_active_search(), "active_tools": tools.list_all(), "available_models": models.list_all(), } logger.info("Tool registered: list_active_tools") @mcp.tool() def health_check() -> Dict[str, str]: """ Health check endpoint for HuggingFace Spaces and monitoring systems. Returns: Dict with service status. """ return {"status": "ok", "service": "Universal MCP Hub"} logger.info("Tool registered: health_check") @mcp.tool() def get_model_info(model_name: str) -> Dict[str, Any]: """ Get limits, costs, and capabilities for a specific model. Args: model_name: Model name as defined in .pyfun [MODELS] (e.g. 'claude-sonnet-4-6'). Returns: Dict with context size, max output tokens, rate limits, costs, and capabilities. Returns empty dict if model is not configured in .pyfun. """ return models.get(model_name) logger.info("Tool registered: get_model_info") # ============================================================================= # DB Tools — uncomment when db_sync.py is ready # ============================================================================= # def _register_db_tools(mcp) -> None: # """ # Register internal SQLite query tool. # Uses db_sync.py (app/* internal SQLite) — NOT postgresql.py (Guardian-only)! # Only SELECT queries are permitted — read-only by design. # """ # from . import db_sync # # @mcp.tool() # async def db_query(query: str) -> list: # """ # Execute a read-only SELECT query on the internal hub state database. # Only SELECT statements are allowed — write operations are blocked. # # Args: # query: SQL SELECT statement to execute. # # Returns: # List of result rows as dicts. # """ # return await db_sync.query(query) # # logger.info("Tool registered: db_query") # ============================================================================= # Direct execution guard # ============================================================================= if __name__ == '__main__': print("WARNING: Run via main.py → app.py, not directly.")