MCPX MCP Gateway

from typing import Dict
import aiohttp

from dataclasses import dataclass
from multidict import CIMultiDictProxy

PROXY_HOST = "sebas-test.lunarhq.dev"
BASE64_PATH = "/base64/{value}"
ANYTHING_PATH = "/anything"
PATH_FACT = "/fact"
HTTPBIN_HOST = "httpbin.org"
CATFACT_HOST = "catfact.ninja"

HTTP_PORT = 80
HTTPS_PORT = 443

GET = "GET"
POST = "POST"

HTTP = "http"
HTTPS = "https"


@dataclass
class ResponseData:
    body: str
    headers: CIMultiDictProxy[str]
    status: int


async def make_request(
    method: str,
    url: str,
    host_header: str,
    body: str | None = None,
    headers: Dict[str, str] | None = None,
) -> ResponseData:
    if headers is None:
        headers = {}
    async with aiohttp.ClientSession() as client:
        async with client.request(
            method,
            url,
            headers={"Host": host_header, **headers},
            data=body,
            allow_redirects=False,
        ) as resp:
            response_headers = resp.headers
            response_body = await resp.text()

            return ResponseData(
                body=response_body,
                headers=response_headers,
                status=resp.status,
            )


def build_url(scheme: str, host: str, path: str):
    return f"{scheme}://{host}{path}"