The Tools Factory module provides functions for initializing and managing Model Context Protocol (MCP) tools.
Lazy initialization of MCP tools - only creates when first needed.
MultiMCPTools instance if URLs configured, empty list otherwise
Behavior
- Returns cached instance if already initialized
- Creates
MultiMCPTools from MCP_URLS environment variable
- Supports comma-separated URLs
- Falls back to default URL if
MCP_URLS not set
- Returns empty list if no URLs configured
Default URL
If MCP_URLS is not set, uses:
https://litellm1-production-4090.up.railway.app/mcp/
Example
from tools.tools_factory import get_mcp_tools
mcp = get_mcp_tools()
if mcp:
print(f"MCP tools available: {type(mcp)}")
else:
print("No MCP tools configured")
setup_mcp
Connect to MCP tools if they exist and haven’t been connected yet.
This function:
- Gets MCP tools instance (lazy initialization)
- Checks if already connected
- Calls
connect() on MultiMCPTools if needed
- Logs success or warning on failure
Example
import asyncio
from tools.tools_factory import setup_mcp
# At application startup
await setup_mcp()
Configuration
MCP_URLS Environment Variable
Comma-separated list of MCP server URLs:
MCP_URLS=https://mcp1.example.com,https://mcp2.example.com,https://mcp3.example.com
Single URL
MCP_URLS=https://mcp.example.com
No Configuration
If MCP_URLS is not set or empty, uses default Railway URL.
The factory creates a MultiMCPTools instance with:
MultiMCPTools(
urls=["url1", "url2", ...],
urls_transports=["streamable-http"]
)
- Transport: Uses
streamable-http for all URLs
- Connection: Lazy - call
await setup_mcp() to connect
Usage in Agent Factory
Integration example from agent_factory.py:
from tools.tools_factory import get_mcp_tools, setup_mcp
# Get MCP tools (lazy init)
mcp_tools = get_mcp_tools()
# Add to agent if available
if mcp_tools:
mcp_agent = Agent(
name="MCP Tools Agent",
model=model,
tools=[mcp_tools],
instructions="You specialize in MCP-based tool interactions."
)
agents.append(mcp_agent)
Usage in Chat Handler
Setup at bot startup:
from tools.tools_factory import setup_mcp
@bot.event
async def on_ready():
# Connect MCP tools
await setup_mcp()
# Rest of initialization...
Cleanup Example
Graceful shutdown:
from tools.tools_factory import get_mcp_tools
async def shutdown():
mcp = get_mcp_tools()
if mcp:
try:
if hasattr(mcp.close, "__await__"):
await mcp.close()
else:
mcp.close()
except Exception as e:
logger.exception("Error closing MCP tools")
State Management
The module maintains global state:
_mcp_tools = None # Cached MultiMCPTools instance
_mcp_connected = False # Connection status flag
Thread Safety
Not thread-safe by design:
- Call
setup_mcp() once at startup
- Call
get_mcp_tools() from any thread/task
Error Handling
Connection Failures
If MCP connection fails:
Failed to connect MCP tools: <error message>
- Returns empty list
[]
- No error or warning
- Agents check for empty list
Complete Example
import asyncio
import logging
from tools.tools_factory import get_mcp_tools, setup_mcp
logger = logging.getLogger(__name__)
async def main():
# Initialize at startup
await setup_mcp()
# Use in agents
mcp = get_mcp_tools()
if mcp:
logger.info("MCP tools available")
# Add to agent...
else:
logger.warning("No MCP tools configured")
# ... rest of application ...
# Cleanup on shutdown
if mcp and hasattr(mcp, 'close'):
try:
await mcp.close()
except Exception as e:
logger.exception("MCP cleanup error")
asyncio.run(main())
Benefits
- Lazy Loading: MCP tools only created when needed
- Singleton Pattern: Same instance reused across calls
- Graceful Degradation: Application works without MCP
- Flexible Configuration: Environment-based URL configuration
- Connection Tracking: Prevents duplicate connections