> ## Documentation Index
> Fetch the complete documentation index at: https://arkos.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.

# 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.

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

## Architecture

```mermaid theme={null}
graph TB
    subgraph "Tool Module"
        TM[MCPToolManager]
        TM --> C1[MCPClient 1]
        TM --> C2[MCPClient 2]
        TM --> CN[MCPClient N]
    end

    subgraph "Transports"
        C1 --> STDIO[StdioTransport]
        C2 --> HTTP[HTTPTransport]
    end

    subgraph "MCP Servers"
        STDIO --> GC[Google Calendar]
        HTTP --> BS[Brave Search]
    end

    subgraph "Per-User Auth"
        TM --> TS[UserTokenStore]
        TS --> DB[(PostgreSQL)]
    end
```

## Core Components

### MCPClient

Manages a single MCP server connection:

```python theme={null}
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:

```python theme={null}
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:

```python theme={null}
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:

```python theme={null}
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

```python theme={null}
# 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

```python theme={null}
# 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

```python theme={null}
# 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

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

### AuthRequiredError

When a user hasn't authenticated:

```python theme={null}
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:

```python theme={null}
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:

```python theme={null}
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

```yaml theme={null}
# 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

```bash theme={null}
# 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:

```python theme={null}
# 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

```python theme={null}
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

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

### Tool Execution Errors

```python theme={null}
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

```python theme={null}
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**:
   ```bash theme={null}
   npm install -g @your-org/your-mcp-server
   ```

2. **Add configuration**:
   ```yaml theme={null}
   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

<AccordionGroup>
  <Accordion title="MCP server not starting">
    Check that the command exists and has correct permissions:

    ```bash theme={null}
    which npx
    npx -y @anthropic/google-calendar-mcp --help
    ```
  </Accordion>

  <Accordion title="Tool not found">
    Verify the tool is registered:

    ```python theme={null}
    tools = await tool_manager.list_all_tools()
    print(tools.keys())
    ```
  </Accordion>

  <Accordion title="Authentication errors">
    Check user token status:

    ```python theme={null}
    status = tool_manager.get_user_service_status(user_id)
    print(status)
    ```
  </Accordion>
</AccordionGroup>

## Next Steps

<CardGroup cols={2}>
  <Card title="Agent Module" icon="robot" href="/modules/agent/overview">
    Learn about agent orchestration
  </Card>

  <Card title="Base Module" icon="layer-group" href="/modules/base/overview">
    Explore the API server
  </Card>

  <Card title="Memory Module" icon="database" href="/modules/memory/overview">
    Understand memory systems
  </Card>

  <Card title="Development Setup" icon="code" href="/development/setup">
    Set up your environment
  </Card>
</CardGroup>