Connect AI Agents to Real Tools.
Safely and in Production.

import { McpServer } from '@modelcontextprotocol/sdk/server/mcp.js';
import { StdioServerTransport } from '@modelcontextprotocol/sdk/server/stdio.js';
import { z } from 'zod';
import { assertSafeUrl } from './security/ssrf.js';

// 1. Initialize production server with graceful lifecycle
export const server = new McpServer({
  name: 'bighub-mcp-production-server',
  version: '1.0.0'
});

// 2. Register SSRF-hardened HTTP tool with strict input schema
server.tool(
  'fetch_safe_url',
  { url: z.string().url().describe('Public HTTPS URL to retrieve') },
  async ({ url }) => {
    // Blocks private CIDRs (10.0.0.0/8, 127.0.0.0/8, AWS 169.254.169.254)
    await assertSafeUrl(url);
    const res = await fetch(url, { signal: AbortSignal.timeout(5000) });
    return { content: [{ type: 'text', text: await res.text() }] };
  }
);

// 3. Dual transport connection: stdio (desktop) or SSE (Docker/Cloud)
const transport = new StdioServerTransport();
await server.connect(transport);

from mcp.server.fastmcp import FastMCP
from pydantic import BaseModel, Field
from src.security.ssrf import validate_safe_url
import httpx

# 1. Initialize production server with async native transport
mcp = FastMCP(
    "bighub-mcp-production-server",
    dependencies=["pydantic", "httpx"]
)

# 2. Register SSRF-hardened HTTP tool with Pydantic validation
@mcp.tool()
async def fetch_safe_url(
    url: str = Field(..., description="Public HTTPS URL to retrieve")
) -> str:
    """Safely fetch public web data, rejecting internal/private IP targets."""
    await validate_safe_url(url)
    async with httpx.AsyncClient(timeout=5.0) as client:
        response = await client.get(url)
        return response.text

# 3. Boot with unbuffered stdio or SSE transport
if __name__ == "__main__":
    mcp.run(transport="stdio")