[{'Text': '"""Remote MCP server using official MCP SDK with HTTP transport and Auth0 OAuth authentication.\n\nThis server implements RFC 9728 OAuth Protected Resource Metadata and uses the official\nMCP Python SDK with StreamableHTTP transport.\n\nAUTHENTICATION:\n- Auth0 OAuth integration using JWT token validation via JWKS\n- RFC 9728 Protected Resource Metadata endpoint at /.well-known/oauth-protected-resource\n- Required scopes: permissions from Auth0 JWT token claims\n\nCONFIGURATION:\nCredentials can be loaded from either:\n1. Environment variables (AUTH0_DOMAIN, AUTH0_CLIENT_ID, etc.)\n2. infrastructure/auth0-credentials.json (created by auth0-setup.sh)\n\nUsage:\n    python -m point_topic_mcp.server_http\n"""\n\nimport os\nos.environ["MCP_TRANSPORT"] = "http"\n\nimport json\nimport logging\nfrom contextlib import asynccontextmanager\nfrom pathlib import Path\nfrom typing import AsyncIterator\n\nfrom dotenv import load_dotenv\nfrom mcp.server import Server\nfrom mcp.types import TextContent, Tool, CallToolRequest\nfrom starlette.applications import Starlette\nfrom starlette.requests import Request\nfrom starlette.responses import JSONResponse, Response\nfrom starlette.routing import Route\n\nfrom point_topic_mcp.tools import get_all_tools\nfrom point_topic_mcp.auth.auth0_helpers import Auth0TokenVerifier\n\n# Load environment variables\nload_dotenv()\n\nlogger = logging.getLogger(__name__)\n\n# Auth0 Configuration\nAUTH0_DOMAIN = os.getenv("AUTH0_DOMAIN", "point-topic.eu.auth0.com")\nAUTH0_AUDIENCE = os.getenv("AUTH0_AUDIENCE", "https://point-topic-mcp.fastmcp.app/")\nAUTH0_CLIENT_ID = os.getenv("AUTH0_CLIENT_ID")\nAUTH0_CLIENT_SECRET = os.getenv("AUTH0_CLIENT_SECRET")\nMCP_BASE_URL = os.getenv("MCP_BASE_URL", "https://point-topic-mcp.fastmcp.app")\n\n# JWKS URL for token validation\nJWKS_URL = f"https://{AUTH0_DOMAIN}/.well-known/jwks.json"\n\n# Global token verifier instance\n_token_verifier: Auth0TokenVerifier | None = None\n\n\ndef load_auth0_credentials():\n    """Load Auth0 credentials from file or environment variables."""\n    # First try environment variables\n    domain = os.getenv("AUTH0_DOMAIN")\n    if domain and domain != "point-topic.eu.auth0.com":\n        return {\n            "domain": domain,\n            "client_id": os.getenv("AUTH0_CLIENT_ID", ""),\n            "client_secret": os.getenv("AUTH0_CLIENT_SECRET", ""),\n            "audience": os.getenv("AUTH0_AUDIENCE", AUTH0_AUDIENCE),\n            "base_url": os.getenv("MCP_BASE_URL", MCP_BASE_URL),\n        }\n\n    # Try credentials file\n    creds_file = (\n        Path(__file__).parent.parent.parent\n        / "infrastructure"\n        / "auth0-credentials.json"\n    )\n    if creds_file.exists():\n        with open(creds_file) as f:\n            creds = json.load(f)\n        return {\n            "domain": creds.get("auth0_domain", AUTH0_DOMAIN),\n            "client_id": creds.get("client_id", AUTH0_CLIENT_ID or ""),\n            "client_secret": creds.get("client_secret", AUTH0_CLIENT_SECRET or ""),\n            "audience": creds.get("api_audience", AUTH0_AUDIENCE),\n            "base_url": os.getenv("MCP_BASE_URL", MCP_BASE_URL),\n        }\n\n    # Return defaults\n    return {\n        "domain": AUTH0_DOMAIN,\n        "client_id": AUTH0_CLIENT_ID or "",\n        "client_secret": AUTH0_CLIENT_SECRET or "",\n        "audience": AUTH0_AUDIENCE,\n        "base_url": MCP_BASE_URL,\n    }\n\n\nasync def oauth_metadata_endpoint(request: Request) -> JSONResponse:\n    """RFC 9728 OAuth Protected Resource Metadata endpoint."""\n    creds = load_auth0_credentials()\n    return JSONResponse(\n        content={\n            "resource": creds["base_url"],\n            "authorization_servers": [f"https://{creds[\'domain\']}/"],\n            "scopes_supported": [],\n            "bearer_methods_supported": ["header"],\n        }\n    )\n\n\nasync def health_endpoint(request: Request) -> JSONResponse:\n    """Health check endpoint."""\n    return JSONResponse(content={"status": "healthy", "server": "Point Topic MCP"})\n\n\nasync def mcp_endpoint(request: Request) -> Response:\n    """MCP endpoint - handles JSON-RPC requests with authentication."""\n    global _token_verifier\n\n    # Validate authentication\n    auth_header = request.headers.get("Authorization", "")\n    if not auth_header.startswith("Bearer "):\n        # Return 401 with WWW-Authenticate header (RFC 9728 compliant)\n        # The resource_metadata parameter tells clients where to find OAuth server info\n        www_auth = (\n            f"Bearer "\n            f\'resource_metadata="{MCP_BASE_URL}/.well-known/oauth-protected-resource"\'\n        )\n        return JSONResponse(\n            status_code=401,\n            headers={\n                "WWW-Authenticate": www_auth,\n                "Content-Type": "application/json",\n            },\n            content={\n                "error": "authentication_required",\n                "error_description": "Valid JWT token required. Use resource_metadata to discover authorization server.",\n            },\n        )\n\n    token = auth_header[7:]  # Remove "Bearer " prefix\n\n    # Validate token\n    if _token_verifier:\n        claims = await _token_verifier.verify_token(token)\n        if not claims:\n            # Return 401 with WWW-Authenticate header (RFC 9728 compliant)\n            www_auth = (\n                f"Bearer "\n                f\'resource_metadata="{MCP_BASE_URL}/.well-known/oauth-protected-resource"\'\n            )\n            return JSONResponse(\n                status_code=401,\n                headers={\n                    "WWW-Authenticate": www_auth,\n                    "Content-Type": "application/json",\n                },\n                content={\n                    "error": "invalid_token",\n                    "error_description": "Token validation failed",\n                },\n            )\n\n        # Store claims in request state for tools to access\n        request.state.token_claims = claims\n\n    # For now, return a placeholder response\n    # In a full implementation, this would proxy to the MCP server\n    return JSONResponse(\n        status_code=200,\n        content={"status": "authenticated"},\n    )\n\n\n@asynccontextmanager\nasync def lifespan(app: Starlette) -> AsyncIterator[None]:\n    """Lifespan context manager for startup/shutdown."""\n    logger.info("Starting Point Topic MCP server with HTTP transport")\n    yield\n    logger.info("Shutting down Point Topic MCP server")\n\n\ndef create_starlette_app() -> Starlette:\n    """Create Starlette app with MCP endpoints."""\n    routes = [\n        Route("/.well-known/oauth-protected-resource", oauth_metadata_endpoint),\n        Route("/health", health_endpoint),\n        Route("/mcp", endpoint=mcp_endpoint, methods=["GET", "POST", "DELETE"]),\n    ]\n\n    return Starlette(debug=True, routes=routes, lifespan=lifespan)\n\n\ndef create_mcp_server() -> Server:\n    """Create and configure MCP server with all tools."""\n    server = Server(name="Point Topic MCP", version="0.2.0")\n\n    # Get all tool functions (returns list of (prefixed_name, func) tuples)\n    tool_functions = get_all_tools()\n    tool_lookup = dict(tool_functions)\n\n    @server.list_tools()\n    async def list_tools() -> list[Tool]:\n        """List available tools."""\n        tools: list[Tool] = []\n\n        for name, func in tool_lookup.items():\n            import inspect\n\n            sig = inspect.signature(func)\n\n            # Build input schema from function signature\n            properties: dict[str, dict[str, str]] = {}\n            required: list[str] = []\n\n            for param_name, param in sig.parameters.items():\n                if param_name.startswith("_"):\n                    continue\n\n                # Map Python types to JSON schema\n                param_type = param.annotation\n                if param_type is str or param_type == inspect.Parameter.empty:\n                    json_type = "string"\n                elif param_type is int:\n                    json_type = "integer"\n                elif param_type is float:\n                    json_type = "number"\n                elif param_type is bool:\n                    json_type = "boolean"\n                else:\n                    json_type = "string"\n\n                properties[param_name] = {"type": json_type}\n\n                if param.default is inspect.Parameter.empty:\n                    required.append(param_name)\n\n            tool = Tool(\n                name=name,\n                description=func.__doc__ or f"Tool: {name}",\n                inputSchema={\n                    "type": "object",\n                    "properties": properties,\n                    "required": required,\n                },\n            )\n            tools.append(tool)\n\n        return tools\n\n    @server.call_tool()\n    async def call_tool(request: CallToolRequest) -> list[TextContent]:\n        """Execute a tool by name."""\n        name = request.params.name\n        arguments = request.params.arguments or {}\n\n        try:\n            tool_func = tool_lookup.get(name)\n            if not tool_func:\n                return [\n                    TextContent(\n                        type="text",\n                        text=json.dumps({"error": f"Tool not found: {name}"}),\n                    )\n                ]\n\n            import inspect\n\n            sig = inspect.signature(tool_func)\n\n            # Filter arguments to only include valid parameters\n            valid_kwargs = {\n                k: v\n                for k, v in arguments.items()\n                if k in sig.parameters and not k.startswith("_")\n            }\n\n            # Execute the tool\n            if inspect.iscoroutinefunction(tool_func):\n                result = await tool_func(**valid_kwargs)\n            else:\n                result = tool_func(**valid_kwargs)\n\n            return [TextContent(type="text", text=json.dumps(result, default=str))]\n\n        except Exception as e:\n            logger.error(f"Error executing tool {name}: {e}")\n            return [TextContent(type="text", text=json.dumps({"error": str(e)}))]\n\n    return server\n\n\ndef main():\n    """Main entry point for the HTTP MCP server."""\n    import uvicorn\n\n    # Setup logging\n    logging.basicConfig(\n        level=logging.INFO,\n        format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",\n    )\n\n    # Load credentials and initialize token verifier\n    creds = load_auth0_credentials()\n    global _token_verifier\n    _token_verifier = Auth0TokenVerifier(\n        auth0_domain=creds["domain"],\n        audience=creds["audience"],\n        jwks_url=JWKS_URL,\n    )\n\n    # Create MCP server (tools will be registered)\n    mcp_server = create_mcp_server()\n    logger.info(\n        f"MCP server created with tools: {[name for name, _ in get_all_tools()]}"\n    )\n\n    # Create Starlette app\n    app = create_starlette_app()\n\n    # Get server configuration\n    host = os.getenv("MCP_HTTP_HOST", "0.0.0.0")\n    port = int(os.getenv("MCP_HTTP_PORT", "8000"))\n\n    logger.info(f"Starting Point Topic MCP server on {host}:{port}")\n    logger.info(f"Auth0 Domain: {AUTH0_DOMAIN}")\n    logger.info(f"Audience: {AUTH0_AUDIENCE}")\n    logger.info(f"OAuth Metadata: {MCP_BASE_URL}/.well-known/oauth-protected-resource")\n    logger.info(f"MCP Endpoint: {MCP_BASE_URL}/mcp")\n\n    # Run the server\n    uvicorn.run(app, host=host, port=port)\n\n\nif __name__ == "__main__":\n    main()\n'}]