http_mcp_server.py

#!/usr/bin/env python3
"""
HTTP MCP Server using FastMCP - curl-like HTTP requests with Bearer token support.
Secured with Google OAuth authentication.
Only allows specific emails to call tools.

Install: uv pip install mcp httpx fastmcp
Run:     uv run http_mcp_server.py
"""

import json
import os
import httpx
from fastmcp import FastMCP
from fastmcp.server.auth.providers.google import GoogleProvider
from fastmcp.server.dependencies import get_access_token

# --- Google OAuth Configuration ---
auth_provider = GoogleProvider(
    client_id=os.getenv("GOOGLE_CLIENT_ID"),
    client_secret=os.getenv("GOOGLE_CLIENT_SECRET"),
    base_url=os.getenv("APP_BASE_URL", "http://localhost:8000"),
    required_scopes=[
        "openid",
        "https://www.googleapis.com/auth/userinfo.email",
    ],
)

mcp = FastMCP(name="http-client", auth=auth_provider)

# --- Allowed Emails ---
ALLOWED_EMAILS = set()
allowed_email_text = os.getenv("ALLOWED_EMAILS")
if allowed_email_text:
    for email in allowed_email_text.split(","):
        ALLOWED_EMAILS.add(email.strip())


def require_allowed_email() -> str:
    """
    Checks the authenticated user's email against the allowlist.
    Raises PermissionError if not allowed.
    Returns the email if allowed.
    """
    token = get_access_token()
    email = token.claims.get("email")

    if not email:
        raise PermissionError("No email found in token. Ensure 'userinfo.email' scope is granted.")

    if email not in ALLOWED_EMAILS:
        raise PermissionError(f"Access denied: {email} is not authorized to use this server.")

    return email


# --- Auth Info Tool ---
@mcp.tool
async def get_user_info() -> dict:
    """Returns information about the authenticated Google user."""
    email = require_allowed_email()

    token = get_access_token()
    return {
        "google_id": token.claims.get("sub"),
        "email": email,
        "name": token.claims.get("name"),
        "picture": token.claims.get("picture"),
        "locale": token.claims.get("locale"),
    }


# --- HTTP Request Tools ---
@mcp.tool
async def http_request(
    url: str,
    method: str = "GET",
    bearer_token: str = "",
    body: str = "",
    content_type: str = "application/json",
    extra_headers: str = "",
    timeout: int = 30,
) -> str:
    """
    Make an HTTP request (like curl) with optional Bearer token auth.

    Args:
        url: Full URL including https://
        method: HTTP method - GET, POST, PUT, PATCH, DELETE (default: GET)
        bearer_token: Bearer token for Authorization header (optional)
        body: Request body as JSON string (optional)
        content_type: Content-Type header (default: application/json)
        extra_headers: Extra headers as JSON string e.g. '{"X-Api-Key": "abc"}' (optional)
        timeout: Request timeout in seconds (default: 30)
    """
    require_allowed_email()

    headers: dict[str, str] = {}

    if bearer_token:
        headers["Authorization"] = f"Bearer {bearer_token}"

    if body:
        headers["Content-Type"] = content_type

    if extra_headers:
        try:
            headers.update(json.loads(extra_headers))
        except json.JSONDecodeError:
            return 'Error: extra_headers must be valid JSON e.g. {"Key": "Value"}'

    try:
        async with httpx.AsyncClient(follow_redirects=True, timeout=timeout) as client:
            response = await client.request(
                method=method.upper(),
                url=url,
                headers=headers,
                content=body.encode() if body else None,
            )

        try:
            response_body = json.dumps(response.json(), indent=2)
        except Exception:
            response_body = response.text

        return (
            f"HTTP {response.status_code} {response.reason_phrase}\n"
            f"URL: {response.url}\n"
            f"Content-Type: {response.headers.get('content-type', '')}\n"
            f"\n--- Response Body ---\n{response_body}"
        )

    except httpx.ConnectError as e:
        return f"Connection error: {e}"
    except httpx.TimeoutException:
        return f"Request timed out after {timeout}s"
    except Exception as e:
        return f"Error: {type(e).__name__}: {e}"


@mcp.tool
async def http_get(url: str, bearer_token: str = "") -> str:
    """
    Make a GET request with optional Bearer token.

    Args:
        url: Full URL including https://
        bearer_token: Bearer token for Authorization header (optional)
    """
    require_allowed_email()
    return await http_request(url=url, method="GET", bearer_token=bearer_token)


@mcp.tool
async def http_post(
    url: str,
    body: str,
    bearer_token: str = "",
    content_type: str = "application/json",
) -> str:
    """
    Make a POST request with a body and optional Bearer token.

    Args:
        url: Full URL including https://
        body: Request body as a JSON string
        bearer_token: Bearer token for Authorization header (optional)
        content_type: Content-Type header (default: application/json)
    """
    require_allowed_email()
    return await http_request(
        url=url,
        method="POST",
        bearer_token=bearer_token,
        body=body,
        content_type=content_type,
    )


# --- Run as HTTP server ---
if __name__ == "__main__":
    mcp.run(
        transport="streamable-http",
        host="0.0.0.0",
        port=7860,
    )