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

Install the MCP server package:

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

Add configuration:

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

Restart ARKOS - the server will be auto-discovered

Best Practices

Use stdio for local tools: Better security and isolation
Use HTTP for shared services: Better for multi-user deployments
Handle auth errors gracefully: Show users how to connect services
Set reasonable timeouts: Prevent hanging on slow tools
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

Getting Started

Architecture

Core Modules

Development

​Tool Module Overview

​What is MCP?

​Architecture

​Core Components

​MCPClient

​MCPToolManager

​Transport Types

​Stdio Transport

​HTTP Transport

​MCP Protocol

​Initialization Handshake

​Tool Discovery

​Tool Execution

​Per-User Authentication

​Services Requiring Auth

​AuthRequiredError

​UserTokenStore

​User Service Status

​Configuration

​YAML Configuration

​Environment Variables

​Integration with Agent

​System Prompt Format

​Error Handling

​Connection Errors

​Tool Execution Errors

​Lifecycle Management

​Graceful Shutdown

​Adding New MCP Servers

​Best Practices

​Troubleshooting

​Next Steps

Agent Module

Base Module

Memory Module

Development Setup

Tool Module Overview

What is MCP?

Architecture

Core Components

MCPClient

MCPToolManager

Transport Types

Stdio Transport

HTTP Transport

MCP Protocol

Initialization Handshake

Tool Discovery

Tool Execution

Per-User Authentication

Services Requiring Auth

AuthRequiredError

UserTokenStore

User Service Status

Configuration

YAML Configuration

Environment Variables

Integration with Agent

System Prompt Format

Error Handling

Connection Errors

Tool Execution Errors

Lifecycle Management

Graceful Shutdown

Adding New MCP Servers

Best Practices

Troubleshooting

Next Steps