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, andcalculate_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
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 │ │
│ └───────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────┘
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
- Server implementation 🖥️ 🔧
#!/usr/bin/env python3
"""
Simple MCP Time Server
Provides tools for time operations, timezone conversions, and time calculations.
"""
import asyncio
import json
from datetime import datetime, timezone, timedelta
from typing import Any, Dict, List
import pytz
from mcp.server import Server
from mcp.server.stdio import stdio_server
from 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
- Client implementation 💻
#!/usr/bin/env python3
"""
ContextForge MCP Gateway Client
Demonstrates how to interact with MCP servers through the ContextForge Gateway
"""
import os
import sys
import json
import requests
from typing import Dict, List, Any, Optional
from datetime import datetime
class 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
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"
}
}
- 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
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. |
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
Top comments (0)