Authentication and Security

pip install fastmcp

import asyncio
from fastmcp import Client
from fastmcp.client.transports import StreamableHttpTransport


async def call_mcp_tool():
    mcp_url = "https://<gateway-url>/mcp/<server-name>/server"
    auth_token = "your-token"

    transport = StreamableHttpTransport(
        url=mcp_url,
        headers={"Authorization": f"Bearer {auth_token}"}
    )

    async with Client(transport) as client:
        # List available tools
        tools = await client.list_tools()
        print(f"Available tools: {[t.name for t in tools]}")

        # Call a specific tool
        result = await client.call_tool("tool_name", {"arg1": "value1"})
        print(f"Result: {result}")


asyncio.run(call_mcp_tool())

import httpx
from fastmcp import Client
from fastmcp.client.transports import StreamableHttpTransport


def _make_httpx_client(**kwargs) -> httpx.AsyncClient:
    """Factory that creates an httpx client which reads error response bodies.

    The MCP library uses streaming requests, so response bodies are not
    automatically read. This hook ensures 4xx/5xx bodies are loaded before
    raise_for_status() discards them.
    """
    async def _read_error_body(response: httpx.Response):
        if response.status_code >= 400:
            await response.aread()

    hooks = kwargs.pop("event_hooks", {})
    hooks.setdefault("response", []).append(_read_error_body)
    return httpx.AsyncClient(event_hooks=hooks, **kwargs)


async def call_mcp_with_oauth(
    mcp_url: str,
    user_token: str,
    tool_name: str,
    tool_args: dict
):
    """Call an MCP tool, handling OAuth authorization if needed."""
    transport = StreamableHttpTransport(
        url=mcp_url,
        headers={"Authorization": f"Bearer {user_token}"},
        httpx_client_factory=_make_httpx_client,
    )

    try:
        async with Client(transport) as client:
            result = await client.call_tool(tool_name, tool_args)
            return {"success": True, "result": result}
    except httpx.HTTPStatusError as e:
        if e.response.status_code == 401:
            body = e.response.json()
            if body.get("error", {}).get("type") == "McpAuthRequiredError":
                return {
                    "success": False,
                    "auth_required": True,
                    "authorization_urls": body.get("authorization_urls", {}),
                    "server_names": body.get("server_names", {}),
                    "message": body.get("message", ""),
                }
        raise


async def main():
    result = await call_mcp_with_oauth(
        mcp_url="https://<gateway-url>/mcp/<server-name>/server",
        user_token="user-tfy-token-or-idp-jwt",
        tool_name="get_repositories",
        tool_args={}
    )

    if result.get("auth_required"):
        print(result["message"])
        for server, url in result["authorization_urls"].items():
            print(f"  Authorize {server}: {url}")
        # After user completes OAuth, retry the request
    else:
        print(f"Success: {result['result']}")

# Same code for all MCP servers - Gateway handles auth injection
transport = StreamableHttpTransport(
    url="https://<gateway-url>/mcp/<virtual-mcp-name>/server",
    headers={"Authorization": f"Bearer {user_token}"}
)

async with Client(transport) as client:
    # Web search - Gateway injects the pre-configured API key
    await client.call_tool("web_search", {"query": "AI news"})

    # Gmail - Gateway injects THIS user's OAuth token
    await client.call_tool("search_emails", {"query": "from:boss"})

import asyncio
import os
from dataclasses import dataclass

import httpx
from fastmcp import Client
from fastmcp.client.transports import StreamableHttpTransport


def _make_httpx_client(**kwargs) -> httpx.AsyncClient:
    """Reads 4xx/5xx response bodies before raise_for_status() discards them."""
    async def _read_error_body(response: httpx.Response):
        if response.status_code >= 400:
            await response.aread()

    hooks = kwargs.pop("event_hooks", {})
    hooks.setdefault("response", []).append(_read_error_body)
    return httpx.AsyncClient(event_hooks=hooks, **kwargs)


@dataclass
class MCPConfig:
    url: str
    name: str


class MCPAgent:
    """An agent that can call tools from multiple MCP servers."""

    def __init__(self, auth_token: str):
        self.auth_token = auth_token
        self.mcp_configs: dict[str, MCPConfig] = {}

    def register_mcp_server(self, name: str, url: str):
        self.mcp_configs[name] = MCPConfig(url=url, name=name)

    async def call_tool(self, server_name: str, tool_name: str, tool_args: dict) -> dict:
        if server_name not in self.mcp_configs:
            raise ValueError(f"MCP server '{server_name}' not registered")

        config = self.mcp_configs[server_name]
        transport = StreamableHttpTransport(
            url=config.url,
            headers={"Authorization": f"Bearer {self.auth_token}"},
            httpx_client_factory=_make_httpx_client,
        )

        try:
            async with Client(transport) as client:
                result = await client.call_tool(tool_name, tool_args)
                return {"success": True, "result": result}
        except httpx.HTTPStatusError as e:
            if e.response.status_code == 401:
                body = e.response.json()
                if body.get("error", {}).get("type") == "McpAuthRequiredError":
                    return {
                        "success": False,
                        "auth_required": True,
                        "authorization_urls": body.get("authorization_urls", {}),
                        "server_names": body.get("server_names", {}),
                        "server": server_name,
                        "message": body.get("message", ""),
                    }
            return {"success": False, "error": str(e)}

    async def list_tools(self, server_name: str) -> list[str]:
        if server_name not in self.mcp_configs:
            raise ValueError(f"MCP server '{server_name}' not registered")

        config = self.mcp_configs[server_name]
        transport = StreamableHttpTransport(
            url=config.url,
            headers={"Authorization": f"Bearer {self.auth_token}"},
            httpx_client_factory=_make_httpx_client,
        )

        async with Client(transport) as client:
            tools = await client.list_tools()
            return [tool.name for tool in tools]


async def main():
    # Initialize with user token for per-user access
    # Or use Virtual Account token for shared access
    agent = MCPAgent(auth_token=os.environ["TFY_API_KEY"])

    agent.register_mcp_server(
        name="github",
        url="https://<gateway-url>/mcp/github-integration/server"
    )
    agent.register_mcp_server(
        name="web-search",
        url="https://<gateway-url>/mcp/web-search/server"
    )

    result = await agent.call_tool(
        server_name="github",
        tool_name="list_repositories",
        tool_args={"owner": "truefoundry"}
    )

    if result.get("auth_required"):
        print("OAuth authorization required:")
        for server, url in result["authorization_urls"].items():
            print(f"  Authorize {server}: {url}")
    elif result.get("success"):
        print(f"Result: {result['result']}")
    else:
        print(f"Error: {result.get('error')}")


if __name__ == "__main__":
    asyncio.run(main())