What is MCP Gateway? I tried the OSS ContextForge as an entry point for asking AI about business system data

# MCP Gateway: Centralizing Authentication, Auditing, and Cost Control for AI Agents ## Concept Overview When AI agents reference business systems, consolidating authentication, auditing, and cost control in one place via an MCP Gateway — and walking through the process of converting existing Web APIs to MCP using the OSS implementation ContextForge. --- ## Part 1: Organizing the MCP Gateway Concept ### Background: The Problem with Direct Agent-to-System Access ``` [Current State - Problematic Pattern] AI Agent A ──────────────────► Business System API AI Agent B ──────────────────► Database AI Agent C ──────────────────► Internal Tool │ ├─ Authentication scattered across each agent ├─ No audit trail ├─ Token costs uncontrolled └─ Security policies inconsistently applied ``` ### MCP Gateway Architecture ``` [Target State - Centralized via MCP Gateway] AI Agent A ──┐ AI Agent B ──┼──► MCP Gateway ──► Business System API AI Agent C ──┘ │ ├──► Database │ └──► Internal Tool │ ┌──────▼──────┐ │ Centralized │ │ • Auth/AuthZ │ │ • Audit Log │ │ • Rate Limit │ │ • Cost Track │ └─────────────┘ ``` ### Three Core Concerns #### 1. Authentication & Authorization ```yaml # Conceptual policy definition auth_policy: agent_identity: method: jwt # Agents present JWT tokens issuer: "https://auth.internal/agents" tool_permissions: agent_type_readonly: allowed_tools: - "search_*" - "get_*" denied_tools: - "delete_*" - "update_*" agent_type_operator: allowed_tools: - "*" requires_approval: - "bulk_delete" - "system_config" ``` #### 2. Audit Trail ```python # Audit log structure (conceptual) audit_event = { "timestamp": "2024-01-15T10:30:00Z", "agent_id": "agent-sales-assistant-001", "agent_type": "sales_assistant", "session_id": "sess_abc123", # MCP call details "tool_name": "search_customer_records", "tool_input": { "query": "enterprise customers", "limit": 10 }, # Result metadata (not full data for PII protection) "result_count": 7, "result_summary": "success", # Cost tracking "upstream_api": "crm_api", "latency_ms": 234, "tokens_consumed": 0, # For non-LLM tools # Risk scoring "data_sensitivity": "confidential", "access_risk_score": 0.3 } ``` #### 3. Cost Control ```python # Cost control policy (conceptual) cost_policy = { "per_agent_limits": { "hourly_tool_calls": 1000, "daily_api_cost_usd": 50.0, "concurrent_sessions": 5 }, "per_tool_limits": { "expensive_ml_inference": { "calls_per_minute": 10, "require_justification": True } }, "global_circuit_breaker": { "daily_total_cost_usd": 500.0, "action": "halt_all_agents" } } ``` --- ## Part 2: Converting Existing Web APIs to MCP with ContextForge ### Environment Setup ```bash # Install ContextForge pip install contextforge # Alternatively, install from source git clone https://github.com/contextforge/contextforge cd contextforge pip install -e ".[dev]" # Verify installation contextforge --version # contextforge 0.3.x ``` ### Step 1: Prepare the Target Web API ```python # target_api/main.py # Simple example business API (FastAPI) from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import Optional, List import uvicorn app = FastAPI(title="Customer Management API", version="1.0.0") # Dummy data CUSTOMERS = { "C001": {"id": "C001", "name": "Acme Corp", "tier": "enterprise", "mrr": 50000}, "C002": {"id": "C002", "name": "StartupXYZ", "tier": "startup", "mrr": 2000}, "C003": {"id": "C003", "name": "MegaCorp", "tier": "enterprise", "mrr": 120000}, } class CustomerUpdate(BaseModel): name: Optional[str] = None tier: Optional[str] = None @app.get("/customers") def list_customers(tier: Optional[str] = None, min_mrr: Optional[int] = None): """List customers with optional filters""" customers = list(CUSTOMERS.values()) if tier: customers = [c for c in customers if c["tier"] == tier] if min_mrr: customers = [c for c in customers if c["mrr"] >= min_mrr] return {"customers": customers, "total": len(customers)} @app.get("/customers/{customer_id}") def get_customer(customer_id: str): """Get specific customer details""" if customer_id not in CUSTOMERS: raise HTTPException(status_code=404, detail="Customer not found") return CUSTOMERS[customer_id] @app.put("/customers/{customer_id}") def update_customer(customer_id: str, update: CustomerUpdate): """Update customer information""" if customer_id not in CUSTOMERS: raise HTTPException(status_code=404, detail="Customer not found") if update.name: CUSTOMERS[customer_id]["name"] = update.name if update.tier: CUSTOMERS[customer_id]["tier"] = update.tier return CUSTOMERS[customer_id] @app.get("/customers/{customer_id}/health-score") def get_health_score(customer_id: str): """Calculate customer health score""" if customer_id not in CUSTOMERS: raise HTTPException(status_code=404, detail="Customer not found") customer = CUSTOMERS[customer_id] # Simple scoring logic score = min(100, customer["mrr"] / 1000 * 2) return { "customer_id": customer_id, "health_score": score, "risk_level": "low" if score > 70 else "medium" if score > 40 else "high" } if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=8080) ``` ```bash # Start the API in a separate terminal python target_api/main.py # INFO: Uvicorn running on http://0.0.0.0:8080 ``` ### Step 2: OpenAPI Spec Generation ```bash # Retrieve the auto-generated spec curl http://localhost:8080/openapi.json | python -m json.tool > customer_api_spec.json # Alternatively, save manually from the spec cat customer_api_spec.json ``` ```json { "openapi": "3.0.2", "info": { "title": "Customer Management API", "version": "1.0.0" }, "paths": { "/customers": { "get": { "summary": "List Customers", "operationId": "list_customers_customers_get", "parameters": [ { "name": "tier", "in": "query", "required": false, "schema": {"type": "string"} }, { "name": "min_mrr", "in": "query", "required": false, "schema": {"type": "integer"} } ] } } } } ``` ### Step 3: ContextForge Configuration ```yaml # contextforge.yaml # MCP server configuration file server: name: "customer-management-mcp" version: "1.0.0" description: "MCP server for customer management operations" # Target API settings upstream: base_url: "http://localhost:8080" openapi_spec: "./customer_api_spec.json" timeout_seconds: 30 # Authentication settings auth: type: "api_key" header: "X-API-Key" value: "${UPSTREAM_API_KEY}" # From environment variable # MCP tool definitions # Auto-generate from OpenAPI spec + custom settings tools: auto_generate: true # Per-tool overrides overrides: list_customers_customers_get: # Rename to human-readable name name: "search_customers" description: | Search and filter the customer list. Filterable by customer tier (enterprise/startup/etc.) and minimum MRR value. # Input schema override input_schema: type: object properties: tier: type: string description: "Customer tier filter (enterprise, startup, smb)" enum: ["enterprise", "startup", "smb"] min_mrr: type: integer description: "Minimum Monthly Recurring Revenue (USD)" minimum: 0 # Permission requirements required_permissions: ["customer:read"] get_customer_customers__customer_id__get: name: "get_customer_details" description: "Retrieve detailed information for a specific customer" required_permissions: ["customer:read"] update_customer_customers__customer_id__put: name: "update_customer" description: "Update customer information (name, tier)" required_permissions: ["customer:write"] # High-risk operation: require confirmation require_confirmation: true confirmation_message: | About to update customer {customer_id}. Changes: {update} Please confirm (yes/no): get_health_score_customers__customer_id__health_score_get: name: "get_customer_health_score" description: | Calculate the health score for a customer. Score range: 0-100 (70+: low risk, 40-70: medium risk, below 40: high risk) required_permissions: ["customer:read", "analytics:read"] # Gateway settings gateway: # Rate limiting rate_limiting: enabled: true default_limits: calls_per_minute: 60 calls_per_hour: 500 tool_specific_limits: update_customer: calls_per_minute: 10 # Stricter for write operations # Audit logging audit: enabled: true log_file: "./logs/mcp_audit.jsonl" log_inputs: true log_outputs: true mask_fields: # PII masking - "email" - "phone" - "credit_card" # Caching cache: enabled: true default_ttl_seconds: 300 tool_specific_ttl: search_customers: 60 get_customer_details: 120 update_customer: 0 # No cache for writes get_customer_health_score: 300 # Transport settings transport: type: "stdio" # or "http", "websocket" # For HTTP transport # http: # host: "0.0.0.0" # port: 3000 # path: "/mcp" ``` ### Step 4: Generate and Launch the MCP Server ```bash # Validate configuration contextforge validate contextforge.yaml # ✓ Configuration valid # ✓ OpenAPI spec loaded: 5 endpoints found # ✓ Tool mappings: 4 tools configured # ✓ Auth settings valid # Preview generated tool definitions contextforge tools list --config contextforge.yaml ``` ``` Available MCP Tools: ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ Tool: search_customers Description: Search and filter the customer list. Permissions: customer:read Rate limit: 60/min Cache TTL: 60s Tool: get_customer_details Description: Retrieve detailed information for a specific customer Permissions: customer:read Rate limit: 60/min Cache TTL: 120s Tool: update_customer Description: Update customer information (name, tier) Permissions: customer:write Rate limit: 10/min Requires confirmation: Yes Cache TTL: No cache Tool: get_customer_health_score Description: Calculate health score for a customer. Permissions: customer:read, analytics:read Rate limit: 60/min Cache TTL: 300s ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ ``` ```bash # Start the MCP server contextforge serve --config contextforge.yaml # INFO: Starting MCP server: customer-management-mcp v1.0.0 # INFO: Loaded 4 tools # INFO: Audit logging enabled: ./logs/mcp_audit.jsonl # INFO: Rate limiting enabled # INFO: Server ready (stdio transport) ``` ### Step 5: MCP Client Implementation ```python # mcp_client_test.py # Test calling the MCP server import asyncio import json from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client async def test_mcp_tools(): """Test MCP tools via ContextForge""" server_params = StdioServerParameters( command="contextforge", args=["serve", "--config", "contextforge.yaml"], env={ "UPSTREAM_API_KEY": "test-key-123" } ) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: # Initialize await session.initialize() print("✓ MCP session initialized") # 1. List available tools tools = await session.list_tools() print(f"\n📋 Available tools: {len(tools.tools)}") for tool in tools.tools: print(f" - {tool.name}: {tool.description[:50]}...") # 2. Test search_customers print("\n🔍 Testing: search_customers") result = await session.call_tool( "search_customers", arguments={ "tier": "enterprise" } ) data = json.loads(result.content[0].text) print(f" Result: {data['total']} enterprise customers found") for customer in data['customers']: print(f" - {customer['name']} (MRR: ${customer['mrr']:,})") # 3. Test get_customer_details print("\n👤 Testing: get_customer_details") result = await session.call_tool( "get_customer_details", arguments={"customer_id": "C001"} ) customer = json.loads(result.content[0].text) print(f" Customer: {customer['name']}, Tier: {customer['tier']}") # 4. Test get_customer_health_score print("\n📊 Testing: get_customer_health_score") result = await session.call_tool( "get_customer_health_score", arguments={"customer_id": "C003"} ) health = json.loads(result.content[0].text) print(f" {health['customer_id']} Health Score: {health['health_score']:.1f}") print(f" Risk Level: {health['risk_level']}") # 5. Test rate limiting print("\n⏱️ Testing: Rate limit enforcement") for i in range(3): result = await session.call_tool( "search_customers", arguments={} ) print(f" Call {i+1}: Success") asyncio.run(test_mcp_tools()) ``` ```bash python mcp_client_test.py ``` ``` ✓ MCP session initialized 📋 Available tools: 4 - search_customers: Search and filter the customer list... - get_customer_details: Retrieve detailed information for a... - update_customer: Update customer information (name, tier... - get_customer_health_score: Calculate health score for a cus... 🔍 Testing: search_customers Result: 2 enterprise customers found - Acme Corp (MRR: $50,000) - MegaCorp (MRR: $120,000) 👤 Testing: get_customer_details Customer: Acme Corp, Tier: enterprise 📊 Testing: get_customer_health_score C003 Health Score: 100.0 Risk Level: low ⏱️ Testing: Rate limit enforcement Call 1: Success Call 2: Success Call 3: Success ``` ### Step 6: Checking the Audit Log ```python # check_audit_log.py # Analyze the audit log import json from pathlib import Path from collections import defaultdict from datetime import datetime def analyze_audit_log(log_file: str = "./logs/mcp_audit.jsonl"): """Audit log analysis""" events = [] with open(log_file) as f: for line in f: if line.strip(): events.append(json.loads(line)) print(f"📋 Total audit events: {len(events)}") print(f"{'='*60}") # Events by tool tool_stats = defaultdict(lambda: {"calls": 0, "errors": 0, "total_latency": 0}) for event in events: tool = event.get("tool_name", "unknown") tool_stats[tool]["calls"] += 1 if event.get("status") == "error": tool_stats[tool]["errors"] += 1 tool_stats[tool]["total_latency"] += event.get("latency_ms", 0) print("\n📊 Calls per tool:") for tool, stats in tool_stats.items(): avg_latency = stats["total_latency"] / stats["calls"] if stats["calls"] > 0 else 0 print(f" {tool}:") print(f" Calls: {stats['calls']}, Errors: {stats['errors']}") print(f" Avg Latency: {avg_latency:.1f}ms") # Recent events print("\n🕐 Recent events (last 5):") for event in events[-5:]: ts = event.get("timestamp", "") tool = event.get("tool_name", "") status = event.get("status", "") agent = event.get("agent_id", "unknown") print(f" [{ts}] {tool} by {agent}: {status}") analyze_audit_log() ``` ``` 📋 Total audit events: 8 ============================================================ 📊 Calls per tool: search_customers: Calls: 4, Errors: 0 Avg Latency: 18.3ms get_customer_details: Calls: 1, Errors: 0 Avg Latency: 12.1ms get_customer_health_score: Calls: 1, Errors: 0 Avg Latency: 22.7ms 🕐 Recent events (last 5): [2024-01-15T10:30:01Z] search_customers by test-client: success [2024-01-15T10:30:02Z] get_customer_details by test-client: success [2024-01-15T10:30:03Z] get_customer_health_score by test-client: success [2024-01-15T10:30:04Z] search_customers by test-client: success [2024-01-15T10:30:05Z] search_customers by test-client: success ``` ### Step 7: Integration with AI Agents ```python # agent_with_mcp.py # Example of an AI agent using the MCP server via ContextForge import asyncio from anthropic import Anthropic from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client import json async def run_customer_analysis_agent(query: str): """ Customer analysis agent Performs customer analysis using tools via MCP """ client = Anthropic() server_params = StdioServerParameters( command="contextforge", args=["serve", "--config", "contextforge.yaml"], env={"UPSTREAM_API_KEY": "production-key"} ) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() # Retrieve available tools and convert to Claude format mcp_tools = await session.list_tools() claude_tools = [] for tool in mcp_tools.tools: claude_tools.append({ "name": tool.name, "description": tool.description, "input_schema": tool.inputSchema }) print(f"🤖 Agent started. Available tools: {len(claude_tools)}") print(f"💬 User query: {query}\n") messages = [{"role": "user", "content": query}] # Agentic loop while True: response = client.messages.create( model="claude-opus-4-5", max_tokens=4096, tools=claude_tools, messages=messages ) # If no tool use, we're done if response.stop_reason == "end_turn": final_response = "" for block in response.content: if hasattr(block, "text"): final_response += block.text print(f"🎯 Agent response:\n{final_response}") break # Execute tool calls tool_results = [] for block in response.content: if block.type == "tool_use": print(f"🔧 Tool call: {block.name}") print(f" Input: {json.dumps(block.input, ensure_ascii=False)}") # Execute via MCP try: result = await session.call_tool( block.name, arguments=block.input ) result_content = result.content[0].text print(f" ✓ Result: {result_content[:100]}...") tool_results.append({ "type": "tool_result", "tool_use_id": block.id, "content": result_content }) except Exception as e: print(f" ✗ Error: {e}") tool_results.append({ "type": "tool_result", "tool_use_id": block.id, "content": f"Error: {str(e)}", "is_error": True }) # Continue conversation messages.append({"role": "assistant", "content": response.content}) messages.append({"role": "user", "content": tool_results}) # Run asyncio.run(run_customer_analysis_agent( "Identify all at-risk enterprise customers and provide improvement recommendations" )) ``` ``` 🤖 Agent started. Available tools: 4 💬 User query: Identify all at-risk enterprise customers and provide improvement recommendations 🔧 Tool call: search_customers Input: {"tier": "enterprise"} ✓ Result: {"customers": [{"id": "C001", "name": "Acme Corp"... 🔧 Tool call: get_customer_health_score Input: {"customer_id": "C001"} ✓ Result: {"customer_id": "C001", "health_score": 100.0, "risk_... 🔧 Tool call: get_customer_health_score Input: {"customer_id": "C003"} ✓ Result: {"customer_id": "C003", "health_score": 100.0, "risk_... 🎯 Agent response: Analysis results for enterprise customers: **Current Status:** - Acme Corp (C001): Health Score 100/100 - Low Risk MRR: $50,000/month - MegaCorp (C003): Health Score 100/100 - Low Risk MRR: $120,000/month **Assessment:** Both enterprise customers are currently in excellent health. However, based on MRR levels, here are proactive improvement recommendations: **Acme Corp:** - Upsell to a higher plan to potentially grow MRR from $50K to $80K+ - Schedule a quarterly business review (QBR) to confirm value delivery **MegaCorp:** - As a top revenue customer, assign a dedicated Customer Success Manager - Develop a case study for marketing (with their permission) - Propose an enterprise multi-year contract for pricing stability ``` --- ## Part 3: Production Deployment Considerations ### MCP Gateway Hardening ```python # gateway_middleware.py # Custom middleware for production deployment from contextforge.middleware import BaseMiddleware from contextforge.types import MCPRequest, MCPResponse import hashlib import time from typing import Optional class SecurityMiddleware(BaseMiddleware): """Security enforcement middleware""" async def before_tool_call( self, request: MCPRequest, context: dict ) -> Optional[MCPResponse]: # 1. Verify agent identity agent_id = context.get("agent_id") if not await self.verify_agent_token(context.get("auth_token")): return MCPResponse.error("Authentication failed", code=401) # 2. Input validation if not await self.validate_input_safety(request.tool_input): return MCPResponse.error("Input validation failed", code=400) # 3. Data access policy check if not await self.check_data_policy(agent_id, request.tool_name): return MCPResponse.error("Access denied by data policy", code=403) return None # Proceed async def after_tool_call( self, request: MCPRequest, response: MCPResponse, context: dict ) -> MCPResponse: # 4. Output PII scrubbing response = await self.scrub_pii(response) # 5. Record compliance log await self.log_compliance_event(request, response, context) return response async def validate_input_safety(self, tool_input: dict) -> bool: """Detect prompt injection and suspicious inputs""" suspicious_patterns = [ "ignore previous instructions", "system prompt", "jailbreak", "<script>", "DROP TABLE", ] input_str = str(tool_input).lower() for pattern in suspicious_patterns: if pattern.lower() in input_str: return False return True class CostTrackingMiddleware(BaseMiddleware): """Cost tracking and control middleware""" def __init__(self, budget_store): self.budget_store = budget_store # Cost per tool call (USD) self.tool_costs = { "search_customers": 0.001, "get_customer_details": 0.001, "get_customer_health_score": 0.005, "update_customer": 0.002, } async def before_tool_call(self, request, context): agent_id = context.get("agent_id") tool_cost = self.tool_costs.get(request.tool_name, 0.001) # Check budget remaining = await self.budget_store.get_remaining_budget(agent_id) if remaining < tool_cost: return MCPResponse.error( f"Insufficient budget. Required: ${tool_cost}, " f"Remaining: ${remaining:.4f}", code=429 ) return None async def after_tool_call(self, request, response, context): agent_id = context.get("agent_id") tool_cost = self.tool_costs.get(request.tool_name, 0.001) # Deduct cost await self.budget_store.deduct_budget( agent_id=agent_id, amount=tool_cost, tool_name=request.tool_name, timestamp=time.time() ) return response ``` ### Deployment Configuration ```yaml # docker-compose.yaml version: '3.8' services: mcp-gateway: build: ./mcp-gateway environment: - UPSTREAM_API_KEY=${UPSTREAM_API_KEY} - AUTH_SECRET=${JWT_SECRET} - REDIS_URL=redis://redis:6379 - LOG_LEVEL=INFO volumes: - ./contextforge.yaml:/app/contextforge.yaml - ./logs:/app/logs ports: - "3000:3000" depends_on: - redis - audit-db redis: image: redis:7-alpine # Rate limiting and caching backend audit-db: image: postgres:15 environment: POSTGRES_DB: mcp_audit POSTGRES_USER: audit_user POSTGRES_PASSWORD: ${AUDIT_DB_PASSWORD} volumes: - audit_data:/var/lib/postgresql/data volumes: audit_data: ``` --- ## Summary ``` MCP Gateway Architecture Summary ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ Problem Solution ───────────────────────── ───────────────────────── Auth scattered across Centralized JWT/API key each agent verification in gateway No audit trail JSONL audit log with full call records Uncontrolled API Per-agent/tool budget costs control + circuit breaker Inconsistent input/ Gateway-level validation output validation and PII scrubbing Tight coupling between MCP abstraction layer agents and API specs (OpenAPI → MCP tools) ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ Implementation Steps (ContextForge) ───────────────────────────────────────────────────── 1. Prepare target API (FastAPI/Express/etc.) 2. Export OpenAPI spec 3. Write contextforge.yaml (tool mapping + policies) 4. contextforge serve starts MCP server 5. AI agents connect via MCP protocol 6. All calls pass through gateway centrally ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ ``` The MCP Gateway pattern enables enterprise-grade safety and governance for AI agent deployments while maintaining development agility — existing Web APIs require no modification to become MCP-compatible tools.