{"schemaVersion":"1.0","type":"Article","types":["Article"],"slug":"forge-your-context-orchestrating-the-contextforge-mcp-gateway-with-ibm-project-bob-ewmxb","url":"https://api.zyvop.com/forge-your-context-orchestrating-the-contextforge-mcp-gateway-with-ibm-project-bob-ewmxb","title":"Forge Your Context: Orchestrating the “ContextForge MCP Gateway” with IBM Project Bob","subtitle":null,"tldr":"Building an MCP Gateway & Registry using demonstration based on “ContextForge MCP Gateway” with IBM Project Bob. Introduction The world of Large Language Mo...","keywords":["MCP","mcpprotocol","mcpserver"],"entities":["Alain Airom (Ayrom)","Build Engineer","MCP","mcpprotocol","mcpserver","ZyVOP"],"keyTakeaways":["Building an MCP Gateway & Registry using demonstration based on “ContextForge MCP Gateway” with IBM Project Bob.","Introduction The world of Large Language Models (LLMs) is evolving at a breakneck pace.","As we move beyond simple prompt-response interactions, the need for robust infrastructure to manage, integrate, and secure LLM-powered applications becomes paramount."],"headings":["Introduction","Presenting ContextForge MCP Gateway","Motivations to build this demonstrator","The Implementation at a Glance","System Architecture","Deployment Flexibility","The code implementation","Adaptability and Comparison of Deployment Strategies","Why This Matters for Your “Bob” Project","Conclusion: Bridging the Gap Between Protocols and Production","Links"],"outboundLinks":["https://github.com/aairom/contextforge-demo/blob/main/EXAMPLES.md","https://github.com/IBM/mcp-context-forge?tab=readme-ov-file","https://github.com/aairom/contextforge-demo","https://www.ibm.com/products/bob"],"contentText":"Building an MCP Gateway & Registry using demonstration based on “ContextForge MCP Gateway” with IBM Project Bob. Introduction The world of Large Language Models (LLMs) is evolving at a breakneck pace. As we move beyond simple prompt-response interactions, the need for robust infrastructure to manage, integrate, and secure LLM-powered applications becomes paramount. This is where the Model Context Protocol (MCP) shines, offering a standardized way for LLMs to interact with external tools, resources, and data. But how do you implement an MCP Gateway and Registry in a practical, demonstrable way? Imagine a central hub that acts as a translator, a security guard, and a resource manager for your LLM ecosystem. Today, I want to propose a compelling approach: building a full-fledged MCP Gateway and Registry application using IBM Project Bob. Presenting ContextForge MCP Gateway ContextForge MCP Gateway is a feature-rich gateway, proxy and MCP Registry that federates MCP and REST services — unifying discovery, auth, rate-limiting, observability, virtual servers, multi-transport protocols, and an optional Admin UI into one clean endpoint for your AI clients. It runs as a fully compliant MCP server, deployable via PyPI or Docker, and scales to multi-cluster environments on Kubernetes with Redis-backed federation and caching. It currently supports: Federation across multiple MCP and REST services A2A (Agent-to-Agent) integration for external AI agents (OpenAI, Anthropic, custom) gRPC-to-MCP translation via automatic reflection-based service discovery Virtualization of legacy APIs as MCP-compliant tools and servers Transport over HTTP, JSON-RPC, WebSocket, SSE (with configurable keepalive), stdio and streamable-HTTP An Admin UI for real-time management, configuration, and log monitoring (with airgapped deployment support) Built-in auth, retries, and rate-limiting with user-scoped OAuth tokens and unconditional X-Upstream-Authorization header support OpenTelemetry observability with Phoenix, Jaeger, Zipkin, and other OTLP backends Scalable deployments via Docker or PyPI, Redis-backed caching, and multi-cluster federation Motivations to build this demonstrator I have long sought a compelling way to showcase the full potential of the ContextForge MCP Gateway through a practical, real-world application. Leveraging IBM Project Bob as my expert development partner, I tasked him with analyzing the gateway’s architecture to design a comprehensive, end-to-end demonstration. The goal was to create a clear, functional proof-of-concept that highlights its unique orchestration and translation capabilities. The result is the following demonstration application, architected by Bob, which bridges the gap between raw MCP protocols and production-ready implementation. The Implementation at a Glance The proposed application follows the architecture defined in the mcp-context-forge repository. It utilizes a layered approach to manage LLM tools securely: The Client Layer: Uses Python and the requests library to interact with the gateway via a REST API, secured by Bearer Token Authentication. The ContextForge Gateway: Acts as the central nervous system, managing a Registry Service, a Virtual Server Manager, and an Admin UI. The Protocol Bridge: The gateway performs the heavy lifting of translating standard REST/SSE calls from the client into the stdio/SSE transport protocols required by MCP servers. The MCP Time Server: A specialized Python-based server that demonstrates the toolkit by providing functions like get_current_time, convert_timezone, and calculate_time_diff. contextforge-demo/ ├── README.md ├── requirements.txt # Python dependencies ├── setup.sh # Setup script ├── run.sh # Run script ├── .env.example # Environment variables template ├── mcp-server/ # Simple MCP time server │ ├── server.py │ └── requirements.txt ├── client/ # Python client application │ ├── client.py │ └── requirements.txt └── docs/ # Additional documentation └── architecture.md Enter fullscreen mode Exit fullscreen mode System Architecture ───────────────────────────────────────────────────────────────┐ │ Client Application │ │ (Python/HTTP) │ └────────────────────────┬─────────────────────────────────────┘ │ │ HTTP REST API / SSE │ Bearer Token Authentication │ ┌────────────────────────▼────────────────────────────────────┐ │ ContextForge MCP Gateway │ │ │ │ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ │ │ Registry │ │ REST API │ │ Admin UI │ │ │ │ Service │ │ Server │ │ (Web) │ │ │ └──────────────┘ └──────────────┘ └──────────────┘ │ │ │ │ ┌──────────────────────────────────────────────────────┐ │ │ │ Virtual Server Manager │ │ │ │ - Server Registration │ │ │ │ - Protocol Translation (MCP ↔ REST/SSE) │ │ │ │ - Tool Catalog Management │ │ │ └───────────────────────────────────────────────────────┘ │ └────────────────────────┬────────────────────────────────────┘ │ │ MCP Protocol (stdio/SSE) │ ┌────────────────────────▼────────────────────────────────────┐ │ MCP Time Server │ │ │ │ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ │ │ Tool Handler │ │ Tool Handler │ │ Tool Handler │ │ │ │ get_time │ │ convert_tz │ │ calc_diff │ │ │ └──────────────┘ └──────────────┘ └──────────────┘ │ │ │ │ ┌───────────────────────────────────────────────────────┐ │ │ │ MCP Server Framework │ │ │ │ - Tool Registration │ │ │ │ - Request/Response Handling │ │ │ │ - stdio Transport │ │ │ └───────────────────────────────────────────────────────┘ │ └─────────────────────────────────────────────────────────────┘ Enter fullscreen mode Exit fullscreen mode Deployment Flexibility Bob’s proposed implementation ensures that the demonstration is accessible across different environments: Local Development: Rapid testing using a simple setup script. Docker: Containerized isolation for consistent environments. Kubernetes: A production-grade blueprint featuring High Availability (HA), auto-scaling, and Redis-backed caching for performance. The code implementation The application implemented, showcased the following main components; Your Computer │ ├─ MCP Time Server (Port 8003) │ └─ Provides 5 time-related tools │ ├─ ContextForge Gateway (Port 4444) │ └─ Manages servers and exposes REST API │ └─ Client Application └─ Calls tools through the gateway Enter fullscreen mode Exit fullscreen mode Server implementation 🖥️ 🔧 #!/usr/bin/env python3 \"\"\" Simple MCP Time Server Provides tools for time operations, timezone conversions, and time calculations. \"\"\" import asyncioimport jsonfrom datetime import datetime, timezone, timedeltafrom typing import Any, Dict, Listimport pytz from mcp.server import Serverfrom mcp.server.stdio import stdio_serverfrom mcp.types import ( Tool, TextContent, ImageContent, EmbeddedResource,)# Initialize the MCP server app = Server(\"time-server\")@app.list_tools()async def list_tools() -> List[Tool]: \"\"\"List available tools.\"\"\" return [ Tool( name=\"get_current_time\", description=\"Get the current time in various formats and timezones\", inputSchema={ \"type\": \"object\", \"properties\": { \"format\": { \"type\": \"string\", \"description\": \"Output format: 'iso', 'unix', 'readable', or 'custom'\", \"enum\": [\"iso\", \"unix\", \"readable\", \"custom\"], \"default\": \"iso\" }, \"timezone\": { \"type\": \"string\", \"description\": \"Timezone name (e.g., 'UTC', 'America/New_York', 'Europe/London')\", \"default\": \"UTC\" }, \"custom_format\": { \"type\": \"string\", \"description\": \"Custom strftime format string (only used when format='custom')\", \"default\": \"%Y-%m-%d %H:%M:%S\" } }, \"required\": [] } ), Tool( name=\"convert_timezone\", description=\"Convert a time from one timezone to another\", inputSchema={ \"type\": \"object\", \"properties\": { \"time_string\": { \"type\": \"string\", \"description\": \"Time string in ISO format (e.g., '2024-01-15T10:30:00')\" }, \"from_timezone\": { \"type\": \"string\", \"description\": \"Source timezone (e.g., 'UTC', 'America/New_York')\", \"default\": \"UTC\" }, \"to_timezone\": { \"type\": \"string\", \"description\": \"Target timezone (e.g., 'Europe/Paris', 'Asia/Tokyo')\" } }, \"required\": [\"time_string\", \"to_timezone\"] } ), Tool( name=\"calculate_time_diff\", description=\"Calculate the difference between two times\", inputSchema={ \"type\": \"object\", \"properties\": { \"start_time\": { \"type\": \"string\", \"description\": \"Start time in ISO format\" }, \"end_time\": { \"type\": \"string\", \"description\": \"End time in ISO format\" }, \"unit\": { \"type\": \"string\", \"description\": \"Unit for the result: 'seconds', 'minutes', 'hours', 'days'\", \"enum\": [\"seconds\", \"minutes\", \"hours\", \"days\"], \"default\": \"hours\" } }, \"required\": [\"start_time\", \"end_time\"] } ), Tool( name=\"list_timezones\", description=\"List all available timezone names, optionally filtered by region\", inputSchema={ \"type\": \"object\", \"properties\": { \"region\": { \"type\": \"string\", \"description\": \"Filter by region (e.g., 'America', 'Europe', 'Asia')\", \"default\": \"\" } }, \"required\": [] } ), Tool( name=\"add_time\", description=\"Add a duration to a given time\", inputSchema={ \"type\": \"object\", \"properties\": { \"base_time\": { \"type\": \"string\", \"description\": \"Base time in ISO format (or 'now' for current time)\" }, \"days\": { \"type\": \"integer\", \"description\": \"Number of days to add\", \"default\": 0 }, \"hours\": { \"type\": \"integer\", \"description\": \"Number of hours to add\", \"default\": 0 }, \"minutes\": { \"type\": \"integer\", \"description\": \"Number of minutes to add\", \"default\": 0 }, \"seconds\": { \"type\": \"integer\", \"description\": \"Number of seconds to add\", \"default\": 0 } }, \"required\": [\"base_time\"] } ) ]@app.call_tool()async def call_tool(name: str, arguments: Dict[str, Any]) -> List[TextContent]: \"\"\"Handle tool calls.\"\"\" try: if name == \"get_current_time\": return await get_current_time(arguments) elif name == \"convert_timezone\": return await convert_timezone(arguments) elif name == \"calculate_time_diff\": return await calculate_time_diff(arguments) elif name == \"list_timezones\": return await list_timezones(arguments) elif name == \"add_time\": return await add_time(arguments) else: return [TextContent( type=\"text\", text=f\"Unknown tool: {name}\" )] except Exception as e: return [TextContent( type=\"text\", text=f\"Error executing tool '{name}': {str(e)}\" )]async def get_current_time(arguments: Dict[str, Any]) -> List[TextContent]: \"\"\"Get current time in specified format and timezone.\"\"\" format_type = arguments.get(\"format\", \"iso\") tz_name = arguments.get(\"timezone\", \"UTC\") custom_format = arguments.get(\"custom_format\", \"%Y-%m-%d %H:%M:%S\") try: tz = pytz.timezone(tz_name) now = datetime.now(tz) if format_type == \"iso\": result = now.isoformat() elif format_type == \"unix\": result = str(int(now.timestamp())) elif format_type == \"readable\": result = now.strftime(\"%A, %B %d, %Y at %I:%M:%S %p %Z\") elif format_type == \"custom\": result = now.strftime(custom_format) else: result = now.isoformat() return [TextContent( type=\"text\", text=json.dumps({ \"time\": result, \"timezone\": tz_name, \"format\": format_type, \"utc_offset\": now.strftime(\"%z\") }, indent=2) )] except Exception as e: return [TextContent( type=\"text\", text=f\"Error: {str(e)}\" )]async def convert_timezone(arguments: Dict[str, Any]) -> List[TextContent]: \"\"\"Convert time between timezones.\"\"\" time_string = arguments.get(\"time_string\") from_tz_name = arguments.get(\"from_timezone\", \"UTC\") to_tz_name = arguments.get(\"to_timezone\") try: from_tz = pytz.timezone(from_tz_name) to_tz = pytz.timezone(to_tz_name) # Parse the time string dt = datetime.fromisoformat(time_string.replace('Z', '+00:00')) # If naive, localize to from_timezone if dt.tzinfo is None: dt = from_tz.localize(dt) # Convert to target timezone converted = dt.astimezone(to_tz) return [TextContent( type=\"text\", text=json.dumps({ \"original_time\": time_string, \"original_timezone\": from_tz_name, \"converted_time\": converted.isoformat(), \"target_timezone\": to_tz_name, \"readable\": converted.strftime(\"%A, %B %d, %Y at %I:%M:%S %p %Z\") }, indent=2) )] except Exception as e: return [TextContent( type=\"text\", text=f\"Error: {str(e)}\" )]async def calculate_time_diff(arguments: Dict[str, Any]) -> List[TextContent]: \"\"\"Calculate difference between two times.\"\"\" start_time = arguments.get(\"start_time\") end_time = arguments.get(\"end_time\") unit = arguments.get(\"unit\", \"hours\") try: start_dt = datetime.fromisoformat(start_time.replace('Z', '+00:00')) end_dt = datetime.fromisoformat(end_time.replace('Z', '+00:00')) diff = end_dt - start_dt total_seconds = diff.total_seconds() if unit == \"seconds\": result = total_seconds elif unit == \"minutes\": result = total_seconds / 60 elif unit == \"hours\": result = total_seconds / 3600 elif unit == \"days\": result = total_seconds / 86400 else: result = total_seconds return [TextContent( type=\"text\", text=json.dumps({ \"start_time\": start_time, \"end_time\": end_time, \"difference\": result, \"unit\": unit, \"human_readable\": str(diff) }, indent=2) )] except Exception as e: return [TextContent( type=\"text\", text=f\"Error: {str(e)}\" )]async def list_timezones(arguments: Dict[str, Any]) -> List[TextContent]: \"\"\"List available timezones.\"\"\" region = arguments.get(\"region\", \"\") try: all_timezones = pytz.all_timezones if region: filtered = [tz for tz in all_timezones if tz.startswith(region)] else: filtered = list(all_timezones) return [TextContent( type=\"text\", text=json.dumps({ \"total_count\": len(filtered), \"region_filter\": region if region else \"none\", \"timezones\": filtered[:50], # Limit to first 50 \"note\": \"Showing first 50 results\" if len(filtered) > 50 else \"All results shown\" }, indent=2) )] except Exception as e: return [TextContent( type=\"text\", text=f\"Error: {str(e)}\" )]async def add_time(arguments: Dict[str, Any]) -> List[TextContent]: \"\"\"Add duration to a time.\"\"\" base_time = arguments.get(\"base_time\") days = arguments.get(\"days\", 0) hours = arguments.get(\"hours\", 0) minutes = arguments.get(\"minutes\", 0) seconds = arguments.get(\"seconds\", 0) try: if base_time.lower() == \"now\": dt = datetime.now(timezone.utc) else: dt = datetime.fromisoformat(base_time.replace('Z', '+00:00')) delta = timedelta(days=days, hours=hours, minutes=minutes, seconds=seconds) result_dt = dt + delta return [TextContent( type=\"text\", text=json.dumps({ \"base_time\": dt.isoformat(), \"added_duration\": { \"days\": days, \"hours\": hours, \"minutes\": minutes, \"seconds\": seconds }, \"result_time\": result_dt.isoformat(), \"human_readable\": result_dt.strftime(\"%A, %B %d, %Y at %I:%M:%S %p %Z\") }, indent=2) )] except Exception as e: return [TextContent( type=\"text\", text=f\"Error: {str(e)}\" )]async def main(): \"\"\"Run the MCP server.\"\"\" async with stdio_server() as (read_stream, write_stream): await app.run( read_stream, write_stream, app.create_initialization_options() )if __name__ == \"__main__\": print(\"Starting Time MCP Server...\", flush=True) asyncio.run(main()) # Made with Bob Enter fullscreen mode Exit fullscreen mode Client implementation 💻 #!/usr/bin/env python3 \"\"\" ContextForge MCP Gateway Client Demonstrates how to interact with MCP servers through the ContextForge Gateway \"\"\" import osimport sysimport jsonimport requestsfrom typing import Dict, List, Any, Optionalfrom datetime import datetimeclass ContextForgeClient: \"\"\"Client for interacting with ContextForge MCP Gateway.\"\"\" def __init__(self, gateway_url: str, bearer_token: str): \"\"\" Initialize the client. Args: gateway_url: Base URL of the gateway (e.g., http://localhost:4444) bearer_token: Bearer token for authentication \"\"\" self.gateway_url = gateway_url.rstrip('/') self.bearer_token = bearer_token self.headers = { 'Authorization': f'Bearer {bearer_token}', 'Content-Type': 'application/json' } def get_version(self) -> Dict[str, Any]: \"\"\"Get gateway version information.\"\"\" response = requests.get( f'{self.gateway_url}/version', headers=self.headers ) response.raise_for_status() return response.json() def list_gateways(self) -> List[Dict[str, Any]]: \"\"\"List all registered gateways.\"\"\" response = requests.get( f'{self.gateway_url}/gateways', headers=self.headers ) response.raise_for_status() return response.json() def list_servers(self) -> List[Dict[str, Any]]: \"\"\"List all registered MCP servers.\"\"\" response = requests.get( f'{self.gateway_url}/servers', headers=self.headers ) response.raise_for_status() return response.json() def register_server(self, name: str, description: str, transport_type: str = \"stdio\", command: str = None, args: List[str] = None, env: Dict[str, str] = None) -> Dict[str, Any]: \"\"\" Register a new MCP server with the gateway. Args: name: Server name description: Server description transport_type: Transport type (stdio, sse, etc.) command: Command to run the server args: Command arguments env: Environment variables \"\"\" payload = { \"server\": { \"name\": name, \"description\": description } } if transport_type: payload[\"server\"][\"transport_type\"] = transport_type if command: payload[\"server\"][\"command\"] = command if args: payload[\"server\"][\"args\"] = args if env: payload[\"server\"][\"env\"] = env response = requests.post( f'{self.gateway_url}/servers', headers=self.headers, json=payload ) response.raise_for_status() return response.json() def list_tools(self, server_name: Optional[str] = None) -> List[Dict[str, Any]]: \"\"\" List all available tools. Args: server_name: Optional server name to filter tools \"\"\" url = f'{self.gateway_url}/tools' if server_name: url += f'?server={server_name}' response = requests.get(url, headers=self.headers) response.raise_for_status() return response.json() def call_tool(self, tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]: \"\"\" Call a tool through the gateway. Args: tool_name: Name of the tool to call arguments: Tool arguments \"\"\" payload = { \"tool\": tool_name, \"arguments\": arguments } response = requests.post( f'{self.gateway_url}/tools/call', headers=self.headers, json=payload ) response.raise_for_status() return response.json()def print_section(title: str): \"\"\"Print a section header.\"\"\" print(f\"\\n{'='*60}\") print(f\" {title}\") print(f\"{'='*60}\\n\")def print_json(data: Any, indent: int = 2): \"\"\"Pretty print JSON data.\"\"\" print(json.dumps(data, indent=indent))def demo_basic_operations(client: ContextForgeClient): \"\"\"Demonstrate basic gateway operations.\"\"\" print_section(\"1. Gateway Version\") try: version = client.get_version() print_json(version) except Exception as e: print(f\"Error: {e}\") print_section(\"2. List Gateways\") try: gateways = client.list_gateways() print_json(gateways) except Exception as e: print(f\"Error: {e}\") print_section(\"3. List Registered Servers\") try: servers = client.list_servers() print(f\"Found {len(servers)} server(s)\") print_json(servers) except Exception as e: print(f\"Error: {e}\")def demo_server_registration(client: ContextForgeClient): \"\"\"Demonstrate server registration.\"\"\" print_section(\"4. Register Time Server\") try: # Register the time server result = client.register_server( name=\"time_server\", description=\"Time and timezone tools for MCP\", transport_type=\"stdio\", command=\"python3\", args=[\"-m\", \"mcp-server.server\"], env={} ) print(\"Server registered successfully!\") print_json(result) except Exception as e: print(f\"Note: Server may already be registered. Error: {e}\")def demo_tool_operations(client: ContextForgeClient): \"\"\"Demonstrate tool listing and calling.\"\"\" print_section(\"5. List Available Tools\") try: tools = client.list_tools() print(f\"Found {len(tools)} tool(s)\") for tool in tools: print(f\"\\n - {tool.get('name', 'unknown')}: {tool.get('description', 'No description')}\") except Exception as e: print(f\"Error: {e}\") return print_section(\"6. Call Tool: get_current_time\") try: result = client.call_tool( \"get_current_time\", { \"format\": \"readable\", \"timezone\": \"UTC\" } ) print(\"Result:\") print_json(result) except Exception as e: print(f\"Error: {e}\") print_section(\"7. Call Tool: convert_timezone\") try: current_time = datetime.utcnow().isoformat() result = client.call_tool( \"convert_timezone\", { \"time_string\": current_time, \"from_timezone\": \"UTC\", \"to_timezone\": \"America/New_York\" } ) print(\"Result:\") print_json(result) except Exception as e: print(f\"Error: {e}\") print_section(\"8. Call Tool: list_timezones\") try: result = client.call_tool( \"list_timezones\", { \"region\": \"America\" } ) print(\"Result:\") print_json(result) except Exception as e: print(f\"Error: {e}\") print_section(\"9. Call Tool: add_time\") try: result = client.call_tool( \"add_time\", { \"base_time\": \"now\", \"days\": 7, \"hours\": 3, \"minutes\": 30 } ) print(\"Result:\") print_json(result) except Exception as e: print(f\"Error: {e}\")def interactive_mode(client: ContextForgeClient): \"\"\"Run in interactive mode.\"\"\" print_section(\"Interactive Mode\") print(\"Available commands:\") print(\" 1. list-tools - List all available tools\") print(\" 2. call-tool - Call a specific tool\") print(\" 3. list-servers - List registered servers\") print(\" 4. version - Show gateway version\") print(\" 5. quit - Exit interactive mode\") while True: print(\"\\n\" + \"-\"*60) command = input(\"\\nEnter command (1-5): \").strip() if command == \"1\" or command.lower() == \"list-tools\": try: tools = client.list_tools() print(f\"\\nFound {len(tools)} tool(s):\") for i, tool in enumerate(tools, 1): print(f\"{i}. {tool.get('name', 'unknown')}\") print(f\" {tool.get('description', 'No description')}\") except Exception as e: print(f\"Error: {e}\") elif command == \"2\" or command.lower() == \"call-tool\": tool_name = input(\"Enter tool name: \").strip() print(\"Enter arguments as JSON (or press Enter for empty):\") args_str = input().strip() try: arguments = json.loads(args_str) if args_str else {} result = client.call_tool(tool_name, arguments) print(\"\\nResult:\") print_json(result) except json.JSONDecodeError: print(\"Error: Invalid JSON format\") except Exception as e: print(f\"Error: {e}\") elif command == \"3\" or command.lower() == \"list-servers\": try: servers = client.list_servers() print(f\"\\nFound {len(servers)} server(s):\") print_json(servers) except Exception as e: print(f\"Error: {e}\") elif command == \"4\" or command.lower() == \"version\": try: version = client.get_version() print(\"\\nGateway Version:\") print_json(version) except Exception as e: print(f\"Error: {e}\") elif command == \"5\" or command.lower() == \"quit\": print(\"\\nExiting interactive mode...\") break else: print(\"Invalid command. Please enter 1-5.\")def main(): \"\"\"Main entry point.\"\"\" # Get configuration from environment gateway_url = os.getenv('GATEWAY_URL', 'http://localhost:4444') bearer_token = os.getenv('MCPGATEWAY_BEARER_TOKEN') if not bearer_token: print(\"Error: MCPGATEWAY_BEARER_TOKEN environment variable not set\") print(\"\\nTo generate a token, run:\") print(\" export MCPGATEWAY_BEARER_TOKEN=$(python3 -m mcpgateway.utils.create_jwt_token \\\\\") print(\" --username admin@example.com --exp 10000 --secret my-test-key)\") sys.exit(1) print(\"=\"*60) print(\" ContextForge MCP Gateway Client Demo\") print(\"=\"*60) print(f\"\\nGateway URL: {gateway_url}\") print(f\"Token: {bearer_token[:20]}...\") # Initialize client client = ContextForgeClient(gateway_url, bearer_token) # Check if running in demo mode or interactive mode mode = os.getenv('CLIENT_MODE', 'demo') if mode == 'interactive': interactive_mode(client) else: # Run demo demo_basic_operations(client) demo_server_registration(client) demo_tool_operations(client) print_section(\"Demo Complete!\") print(\"To run in interactive mode, set CLIENT_MODE=interactive\") print(\"\\nExample:\") print(\" export CLIENT_MODE=interactive\") print(\" python client.py\")if __name__ == \"__main__\": main() # Made with Bob Enter fullscreen mode Exit fullscreen mode Usage examples 📎 Example 1: Get Current Time (ISO Format) curl -s -X POST \\ -H \"Authorization: Bearer $MCPGATEWAY_BEARER_TOKEN\" \\ -H \"Content-Type: application/json\" \\ -d '{ \"tool\": \"get_current_time\", \"arguments\": { \"format\": \"iso\", \"timezone\": \"UTC\" } }' \\ $GATEWAY_URL/tools/call | jq Response: { \"result\": { \"time\": \"2024-01-15T10:30:45.123456+00:00\", \"timezone\": \"UTC\", \"format\": \"iso\", \"utc_offset\": \"+0000\" } } Example 2: Get Current Time (Readable Format) curl -s -X POST \\ -H \"Authorization: Bearer $MCPGATEWAY_BEARER_TOKEN\" \\ -H \"Content-Type: application/json\" \\ -d '{ \"tool\": \"get_current_time\", \"arguments\": { \"format\": \"readable\", \"timezone\": \"America/New_York\" } }' \\ $GATEWAY_URL/tools/call | jq Response: { \"result\": { \"time\": \"Monday, January 15, 2024 at 05:30:45 AM EST\", \"timezone\": \"America/New_York\", \"format\": \"readable\", \"utc_offset\": \"-0500\" } } Example 3: Convert Timezone curl -s -X POST \\ -H \"Authorization: Bearer $MCPGATEWAY_BEARER_TOKEN\" \\ -H \"Content-Type: application/json\" \\ -d '{ \"tool\": \"convert_timezone\", \"arguments\": { \"time_string\": \"2024-01-15T10:30:00\", \"from_timezone\": \"UTC\", \"to_timezone\": \"Asia/Tokyo\" } }' \\ $GATEWAY_URL/tools/call | jq Response: { \"result\": { \"original_time\": \"2024-01-15T10:30:00\", \"original_timezone\": \"UTC\", \"converted_time\": \"2024-01-15T19:30:00+09:00\", \"target_timezone\": \"Asia/Tokyo\", \"readable\": \"Monday, January 15, 2024 at 07:30:00 PM JST\" } } Example 4: Calculate Time Difference curl -s -X POST \\ -H \"Authorization: Bearer $MCPGATEWAY_BEARER_TOKEN\" \\ -H \"Content-Type: application/json\" \\ -d '{ \"tool\": \"calculate_time_diff\", \"arguments\": { \"start_time\": \"2024-01-15T10:00:00Z\", \"end_time\": \"2024-01-15T15:30:00Z\", \"unit\": \"hours\" } }' \\ $GATEWAY_URL/tools/call | jq Response: { \"result\": { \"start_time\": \"2024-01-15T10:00:00Z\", \"end_time\": \"2024-01-15T15:30:00Z\", \"difference\": 5.5, \"unit\": \"hours\", \"human_readable\": \"5:30:00\" } } Example 5: List Timezones by Region curl -s -X POST \\ -H \"Authorization: Bearer $MCPGATEWAY_BEARER_TOKEN\" \\ -H \"Content-Type: application/json\" \\ -d '{ \"tool\": \"list_timezones\", \"arguments\": { \"region\": \"Europe\" } }' \\ $GATEWAY_URL/tools/call | jq Response: { \"result\": { \"total_count\": 58, \"region_filter\": \"Europe\", \"timezones\": [ \"Europe/Amsterdam\", \"Europe/Athens\", \"Europe/Berlin\", \"Europe/Brussels\", \"Europe/London\", \"Europe/Paris\", \"...\" ], \"note\": \"Showing first 50 results\" } } Example 6: Add Time Duration curl -s -X POST \\ -H \"Authorization: Bearer $MCPGATEWAY_BEARER_TOKEN\" \\ -H \"Content-Type: application/json\" \\ -d '{ \"tool\": \"add_time\", \"arguments\": { \"base_time\": \"now\", \"days\": 7, \"hours\": 3, \"minutes\": 30 } }' \\ $GATEWAY_URL/tools/call | jq Response: { \"result\": { \"base_time\": \"2024-01-15T10:30:00+00:00\", \"added_duration\": { \"days\": 7, \"hours\": 3, \"minutes\": 30, \"seconds\": 0 }, \"result_time\": \"2024-01-22T14:00:00+00:00\", \"human_readable\": \"Monday, January 22, 2024 at 02:00:00 PM UTC\" } } Enter fullscreen mode Exit fullscreen mode Settings ⚙️ # Gateway Configuration MCPGATEWAY_UI_ENABLED=true MCPGATEWAY_ADMIN_API_ENABLED=true PLATFORM_ADMIN_EMAIL=admin@example.com PLATFORM_ADMIN_PASSWORD=changeme PLATFORM_ADMIN_FULL_NAME=\"Platform Administrator\" # Server Configuration BASIC_AUTH_PASSWORD=pass JWT_SECRET_KEY=my-test-key Enter fullscreen mode Exit fullscreen mode Adaptability and Comparison of Deployment Strategies | Feature | **Local Development** | **Docker / Docker Compose** | **Kubernetes (K8s)** | | -------------------- | ------------------------------------------------------------ | ---------------------------------------------------------- | ------------------------------------------------------------ | | **Primary Use Case** | Rapid prototyping and initial testing. | Consistent development and staging environments. | Production-grade, high-availability (HA) enterprise environments. | | **Setup Complexity** | **Low**: Requires running a simple setup script (`setup.sh`). | **Medium**: Requires Docker and building images. | **High**: Involves manifest management, Ingress, and secret configuration. | | **Orchestration** | Manual process management (e.g., `run.sh`). | `docker-compose` manages multi-container lifecycle. | Automated pod management, self-healing, and service discovery. | | **Scalability** | Limited to the local machine resources. | Horizontal scaling via container replicas. | High: Multi-replica deployments with auto-scaling. | | **Caching/State** | Local process-based. | Redis container for distributed state. | Redis-backed caching for high-performance clusters. | | **Authentication** | Basic local configuration. | Environment-variable-based secrets. | Robust Kubernetes Secrets management. | | **Observability** | Console logs (stdout/stderr). | Integrated logging via `gateway.log` and `mcp-server.log`. | Full-stack monitoring, health checks, and metrics collection. | Enter fullscreen mode Exit fullscreen mode Why This Matters for Your “Bob” Project As your “expert development assistant,” IBM Project Bob leverages these strategies differently throughout the development lifecycle: During Design: Bob uses the Local architecture to test protocol translation (MCP ↔ REST) and tool logic quickly. During Security Hardening: Bob uses Docker to ensure that environment variables and JWT tokens are handled securely in isolated containers. During Deployment Planning: Bob acts as an Architect to generate the K8s manifests required for the High Availability (HA) production setup described in the architecture.md. Conclusion: Bridging the Gap Between Protocols and Production The ContextForge MCP Gateway stands as a critical piece of infrastructure for the modern AI stack, solving the “last mile” problem of LLM tool integration by providing a secure, observable, and multi-protocol control plane. While the Model Context Protocol offers a standardized language for tools, ContextForge provides the necessary translation layer — converting REST and SSE requests into MCP commands — while adding enterprise essentials like JWT-based authentication and centralized tool discovery. The demonstration application proposed by IBM Project Bob brings these capabilities to life by moving from theory to a functional, scalable reality. By orchestrating a flow from a Python-based client through a high-availability gateway to a specialized MCP Time Server, Bob’s implementation clearly demonstrates how the gateway manages dynamic server registration and tool execution. Whether deployed locally for rapid prototyping or via Kubernetes for production-grade resilience, this Bob-architected solution proves that with the right gateway, your LLM applications can safely and efficiently interact with any external resource, regardless of the underlying protocol. >>> Thanks for reading <<< Links ContextForge MCP Gateway Repository: https://github.com/IBM/mcp-context-forge?tab=readme-ov-file Repository for this code: https://github.com/aairom/contextforge-demo IBM Project Bob: https://www.ibm.com/products/bob","contentHash":"sha256:57dbc4a5bdb062528c6670b6882e839db9be17b3dc641e5fb2065448a09f07f2","authorName":"Alain Airom (Ayrom)","authorUrl":"https://api.zyvop.com/author/alain","authorSameAs":["https://github.com/aairom","https://www.linkedin.com/in/aairom/"],"category":null,"tags":["MCP","mcpprotocol","mcpserver"],"audience":"Software engineers and developers building applications with MCP","tone":"Professional, build engineer perspective","readingTimeMinutes":15,"wordCount":3329,"faqs":null,"primaryTopic":"MCP","publishedAt":"2026-09-03T11:12:49.841Z","updatedAt":"2026-09-03T11:12:49.841Z","canonicalUrl":"https://dev.to/aairom/forge-your-context-orchestrating-the-contextforge-mcp-gateway-with-ibm-project-bob-3p75"}