Tool Module Overview

MCP-compatible tool system for extensible agent capabilities

Tool Module Overview

The ARKOS Tool Module implements the Model Context Protocol (MCP) for connecting to external tool servers, enabling agents to interact with calendars, search engines, and other services.

What is MCP?

The Model Context Protocol (MCP) is a standard for AI systems to communicate with external tools via JSON-RPC 2.0. ARKOS supports both stdio and HTTP transports.

ARKOS tools follow the MCP specification, allowing seamless integration with any MCP-compatible server.

Architecture

Core Components

MCPClient

Manages a single MCP server connection:

from tool_module.tool_call import MCPClient, MCPServerConfig
from tool_module.transports import StdioTransport

# Create config
config = MCPServerConfig(
    name="google-calendar",
    transport="stdio",
    command="npx",
    args=["-y", "@anthropic/google-calendar-mcp"]
)

# Create transport
transport = StdioTransport(
    command=config.command,
    args=config.args,
    env=config.env
)

# Create and start client
client = MCPClient(config, transport)
await client.start()

# List available tools
tools = await client.list_tools()

# Call a tool
result = await client.call_tool(
    name="create_event",
    arguments={"title": "Meeting", "start": "2024-01-26T14:00:00Z"}
)

MCPToolManager

Coordinates multiple MCP server connections:

from tool_module.tool_call import MCPToolManager
from tool_module.token_store import UserTokenStore

# Configuration from YAML
mcp_config = {
    "google-calendar": {
        "transport": "stdio",
        "command": "npx",
        "args": ["-y", "@anthropic/google-calendar-mcp"]
    },
    "brave-search": {
        "transport": "http",
        "url": "http://localhost:8080"
    }
}

# Initialize with token store for per-user auth
token_store = UserTokenStore(db_url)
tool_manager = MCPToolManager(mcp_config, token_store=token_store)

# Initialize all servers
await tool_manager.initialize_servers()

# List all available tools
all_tools = await tool_manager.list_all_tools()

# Call a tool
result = await tool_manager.call_tool(
    tool_name="search",
    arguments={"query": "weather today"},
    user_id="alice"  # For per-user auth
)

Transport Types

Stdio Transport

Launches MCP servers as subprocesses:

from tool_module.transports import StdioTransport

transport = StdioTransport(
    command="npx",
    args=["-y", "@anthropic/google-calendar-mcp"],
    env={"GOOGLE_TOKEN_PATH": "/path/to/token.json"}
)

await transport.connect()
response = await transport.send_request("tools/list", {})
await transport.close()

HTTP Transport

Connects to MCP servers over HTTP:

from tool_module.transports import HTTPTransport

transport = HTTPTransport(
    url="http://localhost:8080",
    auth_config={"api_key": "..."}
)

await transport.connect()
response = await transport.send_request("tools/call", {
    "name": "search",
    "arguments": {"query": "test"}
})

MCP Protocol

Initialization Handshake

# 1. Send initialize request
init_response = await transport.send_request(
    "initialize",
    {
        "protocolVersion": "2024-11-05",
        "capabilities": {},
        "clientInfo": {"name": "arkos", "version": "1.0.0"}
    }
)

# 2. Send initialized notification
await transport.send_notification("notifications/initialized", {})

Tool Discovery

# List available tools
response = await transport.send_request("tools/list", {})
tools = response["result"]["tools"]

# Each tool has:
# - name: Tool identifier
# - description: What the tool does
# - inputSchema: JSON Schema for parameters

Tool Execution

# Call a tool
response = await transport.send_request(
    "tools/call",
    {
        "name": "create_event",
        "arguments": {
            "title": "Team Meeting",
            "start": "2024-01-26T14:00:00Z",
            "duration": 60
        }
    }
)

result = response["result"]

Per-User Authentication

Some tools (like Google Calendar) require per-user OAuth authentication.

Services Requiring Auth

PER_USER_SERVICES = {
    "google-calendar": {
        "name": "Google Calendar",
        "auth_path": "/auth/google/login",
        "scopes": ["calendar"],
    }
}

AuthRequiredError

When a user hasn’t authenticated:

from tool_module.tool_call import AuthRequiredError

try:
    result = await tool_manager.call_tool(
        "create_event",
        {"title": "Meeting"},
        user_id="alice"
    )
except AuthRequiredError as e:
    # Redirect user to authenticate
    print(f"Please connect: {e.connect_url}")
    # e.to_dict() returns structured error for API responses

UserTokenStore

Manages OAuth tokens per user:

from tool_module.token_store import UserTokenStore

token_store = UserTokenStore(db_url)

# Check if user has token
if token_store.has_token(user_id, "google-calendar"):
    # User is authenticated
    pass

# Write token to file for MCP server
token_store.write_token_file(user_id, "google-calendar", "/path/to/token.json")

User Service Status

Check which services a user has connected:

status = tool_manager.get_user_service_status("alice")
# {
#     "google-calendar": {
#         "connected": True,
#         "name": "Google Calendar",
#         "connect_url": "/auth/google/login?user_id=alice"
#     }
# }

# Get missing services
missing = tool_manager.get_missing_services("alice")
# [{"service": "google-calendar", "name": "Google Calendar", ...}]

Configuration

YAML Configuration

# config_module/config.yaml
mcp_servers:
  google-calendar:
    transport: stdio
    command: npx
    args: ["-y", "@anthropic/google-calendar-mcp"]
    env:
      GOOGLE_CALENDAR_MCP_TOKEN_PATH: "${GOOGLE_CALENDAR_MCP_TOKEN_PATH}"

  brave-search:
    transport: http
    url: "http://localhost:8080"
    auth:
      api_key: "${BRAVE_API_KEY}"

Environment Variables

# Google Calendar OAuth
GOOGLE_OAUTH_CREDENTIALS=/path/to/credentials.json
GOOGLE_CALENDAR_MCP_TOKEN_PATH=/path/to/token.json

# Brave Search API
BRAVE_API_KEY=your-api-key

Integration with Agent

The Agent receives tools on startup:

# In base_module/app.py
@app.on_event("startup")
async def startup():
    if tool_manager:
        await tool_manager.initialize_servers()

        # Cache tools on agent
        agent.available_tools = await tool_manager.list_all_tools()

        # Build system prompt with tool descriptions
        tool_prompt = format_tools_for_system_prompt(agent.available_tools)
        agent.system_prompt = base_prompt + "\n\n" + tool_prompt

System Prompt Format

def format_tools_for_system_prompt(tools: dict) -> str:
    lines = [
        "You have access to the following tools.",
        "Use them when appropriate. Only call tools that are listed below.",
        ""
    ]

    for name, tool in tools.items():
        lines.append(f"Tool name: {name}")
        if tool.description:
            lines.append(f"Description: {tool.description}")
        if tool.input_schema:
            lines.append("Input schema:")
            lines.append(str(tool.input_schema))
        lines.append("")

    return "\n".join(lines)

Error Handling

Connection Errors

try:
    await client.start()
except RuntimeError as e:
    logger.error(f"Failed to start MCP server: {e}")

Tool Execution Errors

try:
    result = await client.call_tool(name, arguments)
except RuntimeError as e:
    logger.error(f"Tool execution failed: {e}")
except ValueError as e:
    logger.error(f"Unknown tool: {e}")

Lifecycle Management

Graceful Shutdown

await tool_manager.shutdown()

# This will:
# 1. Stop all shared clients
# 2. Stop all per-user clients
# 3. Clear the tool registry

Adding New MCP Servers

1. Install the MCP server package:

   npm install -g @your-org/your-mcp-server
   

2. Add configuration:

   mcp_servers:
     your-server:
       transport: stdio
       command: npx
       args: ["-y", "@your-org/your-mcp-server"]
   

3. Restart ARKOS - the server will be auto-discovered

Best Practices

1. Use stdio for local tools: Better security and isolation
2. Use HTTP for shared services: Better for multi-user deployments
3. Handle auth errors gracefully: Show users how to connect services
4. Set reasonable timeouts: Prevent hanging on slow tools
5. Log tool calls: Helps with debugging and monitoring

Troubleshooting

MCP server not starting

Check that the command exists and has correct permissions:

which npx
npx -y @anthropic/google-calendar-mcp --help

Tool not found

Verify the tool is registered:

tools = await tool_manager.list_all_tools()
print(tools.keys())

Authentication errors

Check user token status:

status = tool_manager.get_user_service_status(user_id)
print(status)

Next Steps

Agent Module

Learn about agent orchestration

Base Module

Explore the API server

Memory Module

Understand memory systems

Development Setup

Set up your environment