Using a new “Mode” in IBM Bob to create MCP/Agent utils!
Building Smarter Agents: A Guide to IBM’s “MCP Builder with Agent Utils”
In the evolving landscape of AI, the ability for agents to interact with the real world is a game-changer. IBM’s agent’s util library implemented in Bob IDE, often referred to in the context of the "MCP Builder," provides a robust framework for creating Model Context Protocol (MCP) servers. These servers act as a bridge, allowing AI agents (like Claude or GPT-4 or others… you want it, yon name it 😀) to discover and use external tools to perform complex tasks, such as retrieving real-time weather data or performing unit conversions.
The Foundation: BaseMCP and the Tool Registry
At the heart of any MCP server built with these utilities is the BaseMCP class. This base class handles the heavy lifting of the protocol, including tool registration, request routing, and life-cycle management. By extending BaseMCP, developers can focus entirely on the logic of their tools rather than the underlying communication infrastructure.
To expose a Python method as a tool that an agent can call, you simply use the @tool decorator. This decorator automatically registers the method in the server's tool registry, making it discoverable by any connected MCP client.
Anatomy of an MCP Tool
Creating a tool is straightforward but follows a specific pattern to ensure compatibility and performance. Every tool must be an async method and include a Context object as its first parameter. This Context object provides request-scoped information, such as request IDs and metadata, which is essential for logging and tracing in production environments.
A typical tool implementation looks like this:
-
Decorator:
@toolmarks the function for the AI. -
Signature:
async def get_current_weather(self,ctx: Context, location: str, ...). - Docstring: A comprehensive description that tells the AI when and how to use the tool.
-
Return: A dictionary containing a success flag and the resulting
dataor anerrormessage.
"""
Weather MCP Server - A sample implementation using aicoe-agent-utils
This MCP server provides weather-related tools that can be used by AI agents.
It demonstrates the proper structure and patterns for building MCP servers
with the aicoe-agent-utils library.
"""
import os
import asyncio
from typing import Dict, Any, Optional
from datetime import datetime
import httpx
from dotenv import load_dotenv
# Import from aicoe-agent-utils
# For demonstration without access to the actual library, we use a mock implementation
try:
from aicoe_agent_utils.mcp.base_mcp import BaseMCP, tool, Context
except ImportError:
# Fall back to mock implementation
from mock_agent_utils import BaseMCP, tool, Context
# Load environment variables
load_dotenv()
class WeatherMCPServer(BaseMCP):
"""
Weather MCP Server providing weather information tools.
This server extends BaseMCP and provides tools for:
- Getting current weather for a location
- Getting weather forecast
- Converting temperature units
- Getting weather alerts
"""
def __init__(self):
"""Initialize the Weather MCP Server."""
super().__init__()
self.api_key = os.getenv("WEATHER_API_KEY", "demo_key")
self.base_url = "https://api.openweathermap.org/data/2.5"
@tool
async def get_current_weather(
self,
ctx: Context,
location: str,
units: str = "metric"
) -> Dict[str, Any]:
"""
Get current weather information for a specified location.
Args:
ctx: The context object provided by the MCP framework
location: City name or coordinates (e.g., "London" or "51.5074,-0.1278")
units: Temperature units - "metric" (Celsius), "imperial" (Fahrenheit), or "kelvin"
Returns:
Dictionary containing:
- temperature: Current temperature
- feels_like: Feels like temperature
- humidity: Humidity percentage
- description: Weather description
- wind_speed: Wind speed
- timestamp: Time of data calculation
"""
try:
# In a real implementation, this would call the weather API
# For demo purposes, we'll return mock data
async with httpx.AsyncClient() as client:
# Mock response for demonstration
weather_data = {
"location": location,
"temperature": 22.5 if units == "metric" else 72.5,
"feels_like": 21.0 if units == "metric" else 69.8,
"humidity": 65,
"description": "Partly cloudy",
"wind_speed": 5.2 if units == "metric" else 11.6,
"units": units,
"timestamp": datetime.now().isoformat()
}
return {
"success": True,
"data": weather_data
}
except Exception as e:
return {
"success": False,
"error": str(e)
}
@tool
async def get_weather_forecast(
self,
ctx: Context,
location: str,
days: int = 5,
units: str = "metric"
) -> Dict[str, Any]:
"""
Get weather forecast for a specified location.
Args:
ctx: The context object provided by the MCP framework
location: City name or coordinates
days: Number of days to forecast (1-7)
units: Temperature units - "metric", "imperial", or "kelvin"
Returns:
Dictionary containing:
- location: Location name
- forecast: List of daily forecasts with temperature, conditions, etc.
- units: Temperature units used
"""
try:
if days < 1 or days > 7:
return {
"success": False,
"error": "Days must be between 1 and 7"
}
# Mock forecast data
forecast_data = []
for i in range(days):
forecast_data.append({
"day": i + 1,
"date": datetime.now().isoformat(),
"temp_high": 25 + i if units == "metric" else 77 + i * 1.8,
"temp_low": 15 + i if units == "metric" else 59 + i * 1.8,
"description": "Sunny" if i % 2 == 0 else "Cloudy",
"precipitation_chance": 20 + i * 10
})
return {
"success": True,
"data": {
"location": location,
"forecast": forecast_data,
"units": units
}
}
except Exception as e:
return {
"success": False,
"error": str(e)
}
@tool
async def convert_temperature(
self,
ctx: Context,
temperature: float,
from_unit: str,
to_unit: str
) -> Dict[str, Any]:
"""
Convert temperature between different units.
Args:
ctx: The context object provided by the MCP framework
temperature: Temperature value to convert
from_unit: Source unit - "celsius", "fahrenheit", or "kelvin"
to_unit: Target unit - "celsius", "fahrenheit", or "kelvin"
Returns:
Dictionary containing:
- original_value: Input temperature
- original_unit: Input unit
- converted_value: Converted temperature
- converted_unit: Output unit
"""
try:
# Conversion logic
from_unit = from_unit.lower()
to_unit = to_unit.lower()
# Convert to Celsius first
if from_unit == "celsius":
celsius = temperature
elif from_unit == "fahrenheit":
celsius = (temperature - 32) * 5/9
elif from_unit == "kelvin":
celsius = temperature - 273.15
else:
return {
"success": False,
"error": f"Invalid from_unit: {from_unit}"
}
# Convert from Celsius to target unit
if to_unit == "celsius":
result = celsius
elif to_unit == "fahrenheit":
result = (celsius * 9/5) + 32
elif to_unit == "kelvin":
result = celsius + 273.15
else:
return {
"success": False,
"error": f"Invalid to_unit: {to_unit}"
}
return {
"success": True,
"data": {
"original_value": temperature,
"original_unit": from_unit,
"converted_value": round(result, 2),
"converted_unit": to_unit
}
}
except Exception as e:
return {
"success": False,
"error": str(e)
}
@tool
async def get_weather_alerts(
self,
ctx: Context,
location: str
) -> Dict[str, Any]:
"""
Get weather alerts and warnings for a specified location.
Args:
ctx: The context object provided by the MCP framework
location: City name or coordinates
Returns:
Dictionary containing:
- location: Location name
- alerts: List of active weather alerts
- alert_count: Number of active alerts
"""
try:
# Mock alert data
alerts = [
{
"type": "Heat Advisory",
"severity": "Moderate",
"description": "High temperatures expected",
"start_time": datetime.now().isoformat(),
"end_time": datetime.now().isoformat()
}
]
return {
"success": True,
"data": {
"location": location,
"alerts": alerts,
"alert_count": len(alerts)
}
}
except Exception as e:
return {
"success": False,
"error": str(e)
}
async def test(self):
"""
Test method to verify all tools work correctly.
This method is optional but recommended for local testing.
"""
print("Testing Weather MCP Server...")
print("-" * 50)
# Create a mock context
ctx = Context()
# Test get_current_weather
print("\n1. Testing get_current_weather...")
result = await self.get_current_weather(ctx, "London", "metric")
print(f"Result: {result}")
# Test get_weather_forecast
print("\n2. Testing get_weather_forecast...")
result = await self.get_weather_forecast(ctx, "Paris", 3, "metric")
print(f"Result: {result}")
# Test convert_temperature
print("\n3. Testing convert_temperature...")
result = await self.convert_temperature(ctx, 25, "celsius", "fahrenheit")
print(f"Result: {result}")
# Test get_weather_alerts
print("\n4. Testing get_weather_alerts...")
result = await self.get_weather_alerts(ctx, "New York")
print(f"Result: {result}")
print("\n" + "-" * 50)
print("All tests completed!")
if __name__ == "__main__":
import sys
# Create server instance
mcp = WeatherMCPServer()
# Check if running in test mode
if len(sys.argv) > 1 and sys.argv[1] == "test":
# Run tests
asyncio.run(mcp.test())
else:
# Start the MCP server
print("Starting Weather MCP Server...")
print(f"Server will be available at: http://localhost:{os.getenv('WEATHER_MCP_PORT', 8000)}/mcp")
mcp.run()
# Made with Bob
From Development to Production
One of the standout features of this toolkit is the ease of testing. The framework supports a “test mode” where developers can run their tools locally using a built-in test() method before ever deploying to a live environment. This ensures that the logic is sound and the AI-facing documentation is accurate.
For those without immediate access to internal IBM repositories, the project even includes a mock_agent_utils.py implementation. This mock mimics the real library's interface using FastAPI, allowing developers to build and verify their MCP servers entirely offline or in restricted environments.
"""
Mock implementation of aicoe-agent-utils for demonstration purposes.
This file provides mock implementations of the BaseMCP framework components
to allow the Weather MCP Server to run without requiring access to the actual
aicoe-agent-utils library.
In a real implementation, you would use the actual aicoe-agent-utils library
from: https://github.ibm.com/AI-CoE/aicoe-agent-utils
"""
import asyncio
import inspect
import logging
import os
from datetime import datetime
from functools import wraps
from typing import Any, Callable, Dict, Optional
from fastapi import FastAPI, HTTPException
from fastapi.responses import JSONResponse
import uvicorn
# Configure logging
logging.basicConfig(
level=os.getenv("LOG_LEVEL", "INFO"),
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
class Context:
"""
Mock Context object that provides request-scoped information to tools.
In the real aicoe-agent-utils, this would include additional features
like user authentication, session management, and distributed tracing.
"""
def __init__(self):
self.request_id: str = f"req_{datetime.now().timestamp()}"
self.timestamp: datetime = datetime.now()
self.user_id: Optional[str] = None
self.session_id: Optional[str] = None
self.metadata: Dict[str, Any] = {}
def tool(func: Callable) -> Callable:
"""
Mock @tool decorator that marks methods as MCP tools.
In the real aicoe-agent-utils, this decorator would:
- Extract parameter metadata
- Register the tool in the MCP registry
- Add validation and error handling
- Generate OpenAPI documentation
"""
@wraps(func)
async def wrapper(*args, **kwargs):
return await func(*args, **kwargs)
# Mark the function as a tool
wrapper._is_mcp_tool = True
wrapper._tool_name = func.__name__
wrapper._tool_doc = func.__doc__ or ""
return wrapper
class BaseMCP:
"""
Mock BaseMCP base class for building MCP servers.
In the real aicoe-agent-utils, this class would provide:
- Automatic tool discovery and registration
- Request routing and validation
- Response formatting
- Error handling and logging
- Configuration management
- Integration with FastMCP protocol
"""
def __init__(self):
"""Initialize the MCP server."""
self.app = FastAPI(title="MCP Server")
self.tools: Dict[str, Callable] = {}
self.port = int(os.getenv("WEATHER_MCP_PORT", "8000"))
self.host = os.getenv("HOST", "0.0.0.0")
# Discover and register tools
self._discover_tools()
self._setup_routes()
logger.info("MCP Server initialized")
def _discover_tools(self):
"""
Discover all methods decorated with @tool.
This scans the class for methods with the _is_mcp_tool attribute
and registers them in the tools dictionary.
"""
for name, method in inspect.getmembers(self, predicate=inspect.ismethod):
if hasattr(method, '_is_mcp_tool'):
self.tools[method._tool_name] = method
logger.info(f"Registered tool: {method._tool_name}")
def _setup_routes(self):
"""Set up FastAPI routes for the MCP server."""
@self.app.get("/mcp")
async def health_check():
"""Health check endpoint."""
return {
"status": "ok",
"server": "Weather MCP Server",
"tools": list(self.tools.keys())
}
@self.app.get("/mcp/tools")
async def list_tools():
"""List all available tools."""
tools_info = []
for name, method in self.tools.items():
sig = inspect.signature(method)
params = {}
for param_name, param in sig.parameters.items():
if param_name not in ['self', 'ctx']:
params[param_name] = {
"type": str(param.annotation) if param.annotation != inspect.Parameter.empty else "any",
"required": param.default == inspect.Parameter.empty,
"default": None if param.default == inspect.Parameter.empty else param.default
}
tools_info.append({
"name": name,
"description": method._tool_doc.split('\n')[0].strip(),
"parameters": params
})
return {"tools": tools_info}
@self.app.post("/mcp/tools/{tool_name}")
async def execute_tool(tool_name: str, parameters: Dict[str, Any]):
"""Execute a specific tool."""
if tool_name not in self.tools:
raise HTTPException(status_code=404, detail=f"Tool '{tool_name}' not found")
try:
# Create context
ctx = Context()
# Execute tool
logger.info(f"Executing tool: {tool_name}", extra={"request_id": ctx.request_id})
result = await self.tools[tool_name](ctx, **parameters)
logger.info(f"Tool executed successfully: {tool_name}", extra={"request_id": ctx.request_id})
return result
except TypeError as e:
logger.error(f"Invalid parameters for tool {tool_name}: {e}")
raise HTTPException(status_code=422, detail=f"Invalid parameters: {str(e)}")
except Exception as e:
logger.error(f"Error executing tool {tool_name}: {e}", exc_info=True)
raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")
def run(self):
"""
Start the MCP server.
In the real aicoe-agent-utils, this would integrate with the
FastMCP protocol and provide additional features like:
- WebSocket support
- Server-Sent Events (SSE)
- Authentication
- Rate limiting
"""
logger.info(f"Starting MCP Server on {self.host}:{self.port}")
logger.info(f"Server will be available at: http://{self.host}:{self.port}/mcp")
logger.info(f"Registered tools: {', '.join(self.tools.keys())}")
uvicorn.run(
self.app,
host=self.host,
port=self.port,
log_level=os.getenv("LOG_LEVEL", "info").lower()
)
# Export the mock implementations
__all__ = ['BaseMCP', 'tool', 'Context']
# Made with Bob
Best Practices for Scalable Agents
To build production-ready agents, the documentation emphasizes several best practices:
-
Non-blocking I/O: Always use
async/awaitand asynchronous clients like httpx to prevent the server from hanging during API calls. - Type Hinting: Use Python type hints for all parameters so the MCP framework can validate inputs automatically.
- Structured Logging: Leverage the built-in logger to track tool execution and debug issues in real-time.
- Error Handling: Wrap tool logic in try-except blocks to return clean, standardized error responses to the agent.
python3 weather_mcp_server.py test
2026-03-30 10:42:27,880 - mock_agent_utils - INFO - Registered tool: convert_temperature
2026-03-30 10:42:27,880 - mock_agent_utils - INFO - Registered tool: get_current_weather
2026-03-30 10:42:27,880 - mock_agent_utils - INFO - Registered tool: get_weather_alerts
2026-03-30 10:42:27,880 - mock_agent_utils - INFO - Registered tool: get_weather_forecast
2026-03-30 10:42:27,882 - mock_agent_utils - INFO - MCP Server initialized
Testing Weather MCP Server...
--------------------------------------------------
1. Testing get_current_weather...
Result: {'success': True, 'data': {'location': 'London', 'temperature': 22.5, 'feels_like': 21.0, 'humidity': 65, 'description': 'Partly cloudy', 'wind_speed': 5.2, 'units': 'metric', 'timestamp': '2026-03-30T10:42:27.948120'}}
2. Testing get_weather_forecast...
Result: {'success': True, 'data': {'location': 'Paris', 'forecast': [{'day': 1, 'date': '2026-03-30T10:42:27.951672', 'temp_high': 25, 'temp_low': 15, 'description': 'Sunny', 'precipitation_chance': 20}, {'day': 2, 'date': '2026-03-30T10:42:27.951678', 'temp_high': 26, 'temp_low': 16, 'description': 'Cloudy', 'precipitation_chance': 30}, {'day': 3, 'date': '2026-03-30T10:42:27.951681', 'temp_high': 27, 'temp_low': 17, 'description': 'Sunny', 'precipitation_chance': 40}], 'units': 'metric'}}
3. Testing convert_temperature...
Result: {'success': True, 'data': {'original_value': 25, 'original_unit': 'celsius', 'converted_value': 77.0, 'converted_unit': 'fahrenheit'}}
4. Testing get_weather_alerts...
Result: {'success': True, 'data': {'location': 'New York', 'alerts': [{'type': 'Heat Advisory', 'severity': 'Moderate', 'description': 'High temperatures expected', 'start_time': '2026-03-30T10:42:27.951711', 'end_time': '2026-03-30T10:42:27.951712'}], 'alert_count': 1}}
--------------------------------------------------
All tests completed!
Conclusion
By following these patterns and using Bob (and specific modes), developers can rapidly move from a simple concept to a sophisticated, tool-augmented AI agent capable of navigating complex real-world workflows.
>>> Thanks for reading <<<
Links
- IBM Bob Download: https://bob.ibm.com/
- IBM Bob’s Documentation: https://bob.ibm.com/docs/ide
- Understanding MCP: https://bob.ibm.com/docs/ide/configuration/mcp/understanding-mcp
Top comments (0)