# ============================================================================= # root/app/mcp.py # 14.03.2026 # 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. # # TRANSPORT: # Primary: Streamable HTTP (MCP spec 2025-11-25) → single /mcp endpoint # Configured via HUB_TRANSPORT = "streamable-http" in .pyfun [HUB] # ASGI-App via get_asgi_app() → mounted by app/app.py # # Fallback: SSE (legacy, deprecated per spec) → /mcp route via Quart # Configured via HUB_TRANSPORT = "sse" in .pyfun [HUB] # handle_request() called directly by app/app.py Quart route # # All MCP traffic (both transports) passes through app/app.py first — # auth checks, rate limiting, logging can be added there before reaching MCP. # # 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 contextlib import logging from typing import Dict, Any from . import config as app_config from . import providers from . import models from . import pricing from . import tools logger = logging.getLogger('mcp') # ============================================================================= # Globals — set once during initialize(), never touched elsewhere # ============================================================================= _mcp = None # FastMCP instance _transport = None # "streamable-http" | "sse" — from .pyfun [HUB] HUB_TRANSPORT _stateless = None # True = HF Spaces / horizontal scaling safe # ============================================================================= # 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. Reads HUB_TRANSPORT and HUB_STATELESS from .pyfun [HUB]. Transport modes: streamable-http → get_asgi_app() returns ASGI app → app.py mounts it sse → handle_request() used by Quart route in app.py 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 → db_sync SELECT-only (app/* tables) 5. Persist tools → memory orchestration (works WITHOUT PostgreSQL — queue) """ global _mcp, _transport, _stateless hub_cfg = app_config.get_hub() _transport = hub_cfg.get("HUB_TRANSPORT", "streamable-http").lower() _stateless = hub_cfg.get("HUB_STATELESS", "true").lower() == "true" logger.info(f"MCP Hub initializing (transport: {_transport}, stateless: {_stateless})...") 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." ), stateless_http=_stateless, # True = no session state, HF Spaces safe ) # --- Initialize registries --- providers.initialize() models.initialize() tools.initialize() # --- Register MCP tools --- _register_llm_tools(_mcp) _register_search_tools(_mcp) _register_system_tools(_mcp) _register_pricing_tools(_mcp) _register_db_tools(_mcp) # db_sync ready — SELECT-only, app/* tables _register_persist_tools(_mcp) # memory orchestration — queue-safe, no PG required _register_integration_tools(_mcp) # discord_post / gsc_query — only if active in .pyfun _register_skills_tools(_mcp) # show_skills / search_skills — read-only skill DBs logger.info(f"MCP Hub initialized. Transport: {_transport}") # ============================================================================= # ASGI App — used by app/app.py for Streamable HTTP transport # ============================================================================= def get_asgi_app(): """ Returns the ASGI app for the configured transport. Called by app/app.py AFTER initialize() — mounted as ASGI sub-app. Streamable HTTP: mounts on /mcp — single endpoint for all MCP traffic. SSE (fallback): returns sse_app() for legacy client compatibility. NOTE: For SSE transport, app/app.py uses the Quart route + handle_request() instead — get_asgi_app() is only called for streamable-http. """ if _mcp is None: raise RuntimeError("MCP not initialized — call initialize() first.") if _transport == "streamable-http": logger.info("MCP ASGI app: Streamable HTTP → /mcp") return _mcp.streamable_http_app() else: # SSE as ASGI app — only used if app.py mounts it directly # (normally app.py uses the Quart route + handle_request() for SSE) logger.info("MCP ASGI app: SSE (legacy) → /sse") return _mcp.sse_app() # ============================================================================= # Mount Prefixes — welche URL-Prefixe gehören der MCP-ASGI-App # ============================================================================= def mount_prefixes(): """URL prefixes the mounted ASGI app owns (used by app/app.py).""" if _transport == "streamable-http": return ("/mcp",) return ("/sse", "/messages") # ============================================================================= # Lifespan — Session-Manager für die Lebensdauer des Servers # ============================================================================= @contextlib.asynccontextmanager async def lifespan(): """ Runs the FastMCP session manager for the lifetime of the server. WHY: FastMCP's streamable_http_app() starts its session manager in the Starlette lifespan. Because app/app.py mounts it behind a plain ASGI middleware (lifespan scope goes to Quart, not to the inner app), we have to run it ourselves — otherwise every request fails with 'Task group is not initialized'. Call AFTER get_asgi_app() (it creates the session manager). """ if _mcp is not None and _transport == "streamable-http": async with _mcp.session_manager.run(): yield else: yield # ============================================================================= # 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", ""), "transport": _transport, "active_llm_providers": providers.list_active_llm(), "active_search_providers": providers.list_active_search(), "active_tools": tools.list_all(), "available_models": models.list_all(), "pricing_stats": pricing.stats(), } logger.info("Tool registered: list_active_tools") @mcp.tool() async def health_check() -> Dict[str, Any]: """ Health check: hub status + PostgreSQL awake-check (short ping, also wakes a sleeping Neon), persist queue depth, active providers, integrations. Returns: Dict — status "ok" | "degraded" (PostgreSQL configured but not answering). """ result = await tools.health() result["transport"] = _transport return result logger.info("Tool registered: health_check") @mcp.tool() async def get_model_info(model_name: str) -> Dict[str, Any]: """ Get limits, costs, and capabilities for a specific model. Costs are MERGED: .pyfun manual values win, auto-fetched values (OpenRouter / LiteLLM) fill the gaps. 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 pricing source info. Empty dict if model unknown. """ # .pyfun [MODELS] als Basis + Auto-Pricing füllt Lücken merged = await pricing.get_pricing(model_name) if not merged.get("provider") and not merged.get("cost_input_per_1k"): return merged # komplett unbekanntes Model return merged logger.info("Tool registered: get_model_info") # ============================================================================= # Pricing Tools — auto-fetched model costs (OpenRouter + LiteLLM) # ============================================================================= def _register_pricing_tools(mcp) -> None: """ Register pricing tools. Answers the question nobody wants to maintain by hand: 'Was kostet dieses Model?' — ohne [.MODELS] Kosten von Hand zu pflegen. """ @mcp.tool() async def get_model_pricing(model_name: str) -> Dict[str, Any]: """ Get current pricing for a model — auto-fetched, no manual maintenance. Merge order (highest priority first): 1. .pyfun [MODELS] manual values — your override always wins 2. OpenRouter public models API (all OR models, per-token USD) 3. LiteLLM community price list (Anthropic, Gemini, OpenAI, ...) Args: model_name: Model name (e.g. 'claude-opus-4-6', 'gemini-2.5-flash', 'meta-llama/llama-3.3-70b-instruct'). Returns: Dict with cost_input_per_1k, cost_output_per_1k (USD), context_tokens, max_output_tokens, source, pricing_note. """ return await pricing.get_pricing(model_name) logger.info("Tool registered: get_model_pricing") @mcp.tool() async def refresh_pricing() -> Dict[str, Any]: """ Force a pricing refresh from OpenRouter + LiteLLM. Updates the db_sync cache (TTL-gated, default 24h). Returns: Dict with fetched_at, per-source model counts, and total cached models. """ return await pricing.refresh() logger.info("Tool registered: refresh_pricing") @mcp.tool() async def usage_stats(group_by: str = "model", days: int = 0) -> Dict[str, Any]: """ What did my LLM calls cost? Aggregated from the usage_log table. Args: group_by: "model" | "provider" | "tool" | "day". Default: "model". days: Only the last N days (0 = all time). Returns: Dict with rows (calls, tokens_in, tokens_out, cost_usd, calls_unpriced per group) and a total. cost_usd is an ESTIMATE from merged pricing; calls_unpriced counts calls for which no price was known. """ from . import db_sync return await db_sync.usage_stats(group_by=group_by, days=days or None) logger.info("Tool registered: usage_stats") @mcp.tool() async def openrouter_catalog(query: str = "", free_only: bool = True, limit: int = 40) -> Dict[str, Any]: """ Real model IDs from the live OpenRouter catalog — to pick valid entries for .pyfun [MODELS] (free models come and go; `stale_models` in health_check lists the ones that vanished). Args: query: Substring filter, e.g. "llama", "qwen", "gemma". free_only: Only ":free" models (default True). limit: Max rows (1-200). Returns: {in_catalog, matches, shown, models:[{id, context_tokens, cost_input_per_1k, cost_output_per_1k}]} """ from . import pricing return await pricing.catalog(query=query, free_only=free_only, limit=limit) logger.info("Tool registered: openrouter_catalog") # ============================================================================= # DB Tools — db_sync.py is ready, enabled # ============================================================================= def _register_db_tools(mcp) -> None: """ Register internal SQLite query tool. Uses db_sync.py (app/* internal SQLite) — NOT postgresql.py (Guardian-only)! SECURITY: Only a single SELECT is permitted. Enforced in db_sync.query(): read-only connection, users/sessions/sqlite internals rejected, max 1000 rows. Tables accessible: hub_state, tool_cache, persist_queue, usage_log (app/* only) Tables blocked: users, sessions (Guardian-only, different owner) Requirements (all met): 1. This function enabled 2. _register_db_tools(_mcp) called in initialize() 3. db_sync.initialize() runs in app/app.py BEFORE mcp.initialize() """ from . import db_sync @mcp.tool() async def db_query(sql: str) -> list: """ Execute a read-only SELECT query on the internal hub state database. Only SELECT statements are permitted — all write operations are blocked at the db_sync layer (not just by convention). Accessible tables: hub_state — current hub runtime state (tool status, uptime, etc.) tool_cache — cached tool results for repeated queries persist_queue — memory queue waiting for PostgreSQL usage_log — tokens + estimated cost per LLM call NOT accessible (Guardian-only): users — managed by fundaments/user_handler.py sessions — managed by fundaments/user_handler.py Args: sql: SQL SELECT statement. Example: "SELECT * FROM hub_state LIMIT 10" Returns: List of result rows as dicts. Empty list if no results. Raises: ValueError: If statement is not a SELECT query. RuntimeError: If db_sync is not initialized. """ return await db_sync.query(sql) logger.info("Tool registered: db_query (SQLite SELECT-only, app/* tables)") # ============================================================================= # Persist Tools — Gedächtnis-Orchestrierung (SQLite-Queue → PostgreSQL) # ============================================================================= def _register_skills_tools(mcp) -> None: """show_skills / search_skills — only if active in .pyfun AND skill DBs were found.""" from . import skills_db if not skills_db.is_ready(): return if "show_skills" in tools.list_all(): @mcp.tool() async def show_skills(scope: str = "all", tag: str = "") -> Any: """ List available skills (name, description, tags, chunk count). Args: scope: "all" (default) | "user" | "agent". The core skill is never listed. tag: Optional tag filter. """ return await tools.run("show_skills", scope=scope, tag=tag) logger.info("Tool registered: show_skills") if "search_skills" in tools.list_all(): @mcp.tool() async def search_skills(query: str, top_k: int = 3, tag: str = "", scope: str = "user") -> Any: """ Keyword search (BM25) over skills — returns the best matching chunks. Args: query: Search words. top_k: 1-10 chunks. Default 3. tag: Optional tag filter. scope: "user" (default) | "agent" | "all". The core skill is never searchable. """ return await tools.run("search_skills", prompt=query, top_k=top_k, tag=tag, scope=scope) logger.info("Tool registered: search_skills") def _register_integration_tools(mcp) -> None: """ Optional integrations — registered ONLY if the tool is active in .pyfun AND the module is ready. discord_post → hub posts into Discord (app/discord_api.py, [DISCORD]) gsc_query → Search Console top queries/pages (app/gsc.py, [GSC]) """ if "discord_post" in tools.list_all(): from . import discord_api if discord_api.is_ready(): @mcp.tool() async def discord_post(message: str, channel_id: str = "") -> Dict[str, Any]: """ Post a message to Discord (max 2000 chars). Args: message: The text to post. channel_id: Optional channel — only accepted if listed in DISCORD_ALLOWED_CHANNELS. Empty = default channel / webhook. Returns: {"ok": bool, "error": str|None, "message_id": str|None} """ return await tools.run("discord_post", prompt=message, channel_id=channel_id) logger.info("Tool registered: discord_post") else: logger.info("discord_post active in .pyfun but Discord not ready — skipped.") if "gsc_query" in tools.list_all(): from . import gsc if gsc.is_ready(): @mcp.tool() async def gsc_query( property_url: str = "", days_back: int = 30, row_limit: int = 10, dimension: str = "query", ) -> Any: """ Google Search Console: top queries (or pages) of a property, read-only. Args: property_url: 'sc-domain:example.com' or 'https://example.com/'. Empty = default property. days_back: 1-365 days (data lags 2-3 days). Default 30. row_limit: 1-100 rows. Default 10. dimension: query | page | country | device | date. Default: query. Returns: List of {keyword|page|..., clicks, impressions, ctr (%), position}. """ return await tools.run( "gsc_query", property_url=property_url, days_back=days_back, row_limit=row_limit, dimension=dimension, ) logger.info("Tool registered: gsc_query") else: logger.info("gsc_query active in .pyfun but GSC not ready — skipped.") def _register_persist_tools(mcp) -> None: """ Register memory-orchestration tools. persist_result works WITHOUT PostgreSQL: entries queue restart-safe in the SQLite persist_queue (db_sync) and flush automatically once the Guardian-injected PostgreSQL bridge is active. No key = queued, not crash. """ if "persist_result" not in tools.list_all(): logger.info("persist_result not active in .pyfun — skipped.") return @mcp.tool() async def persist_result( prompt: str = None, source: str = None, model: str = None, score: float = None, ) -> Dict[str, Any]: """ Persist a result to long-term memory (PostgreSQL). Restart-safe: without DATABASE_URL the entry queues in SQLite and flushes automatically once PostgreSQL is available. Never crashes. Args: prompt: The text/result to persist (or hub_state key via .pyfun state_read_key). source: Which tool/component produced this entry (optional). model: Which model produced this entry (optional). score: Optional quality score — reserved slot, set by your own filter. Returns: Dict with status ('persisted' | 'queued'), target_table, queue_depth. """ return await tools.run( tool_name="persist_result", prompt=prompt, source=source, model=model, score=score, ) logger.info("Tool registered: persist_result (queue-safe memory orchestration)") @mcp.tool() async def persist_status() -> Dict[str, Any]: """ Show memory orchestration status: queue depth, oldest entry, failed attempts, and whether the PostgreSQL bridge is active. Returns: Dict with queue stats — observability for the memory pipeline. """ from . import db_sync return await db_sync.persist_queue_stats() logger.info("Tool registered: persist_status") # ============================================================================= # Direct execution guard # ============================================================================= if __name__ == '__main__': print("WARNING: Run via main.py → app.py, not directly.")