Connect Your Tools

Continue from Run Claude or Run Codex. Both agents will use the same HTTP MCP service to read a support ticket and an order, then save an internal note to a disposable SQLite database.

The harness is the MCP client here. This is separate from exposing agents as MCP tools.

Start the business tool service

In the environment used for your agent, install MCP 2.x:

uv pip install "mcp>=2,<3"

Save as support_tools.py:

support_tools.py

import sqlite3
from pathlib import Path

from mcp.server.mcpserver import MCPServer

Path("data").mkdir(exist_ok=True)
mcp = MCPServer("support")

@mcp.tool()
def get_ticket(ticket_id: str) -> dict:
    """Read a ticket from the disposable support fixture."""
    if ticket_id != "T-100":
        raise ValueError("Unknown ticket")
    return {"ticket_id": ticket_id, "order_id": "O-100", "question": "Where is my order?"}

@mcp.tool()
def get_order(order_id: str) -> dict:
    """Read an order from the disposable support fixture."""
    if order_id != "O-100":
        raise ValueError("Unknown order")
    return {"order_id": order_id, "status": "shipped", "delivery": "tomorrow"}

@mcp.tool()
def save_note(operation_id: str, ticket_id: str, note: str) -> dict:
    """Save one internal note per operation ID; repeated IDs return the saved note."""
    if ticket_id != "T-100":
        raise ValueError("Unknown ticket")
    with sqlite3.connect("data/support-tools.db") as db:
        db.execute("CREATE TABLE IF NOT EXISTS notes (operation_id TEXT PRIMARY KEY, ticket_id TEXT, note TEXT)")
        db.execute("INSERT OR IGNORE INTO notes VALUES (?, ?, ?)", (operation_id, ticket_id, note))
        row = db.execute("SELECT ticket_id, note FROM notes WHERE operation_id=?", (operation_id,)).fetchone()
        if row[0] != ticket_id or row[1] != note:
            raise ValueError("Operation ID already used with different arguments")
    return {"operation_id": operation_id, "ticket_id": row[0], "note": row[1]}

if __name__ == "__main__":
    mcp.run(transport="streamable-http", host="127.0.0.1", port=8001)

Run it in a separate terminal with the same environment activated:

python support_tools.py

This unauthenticated fixture binds loopback and changes only data/support-tools.db. Real tools should authenticate callers, enforce tenant access and preserve the same operation-ID discipline.

Connect the harness

Keep the database, working directory and AgentOS setup from your first service. Replace the agent = ... constructor with the matching version below.

Claude

agent = ClaudeAgent(
    id="support-agent",
    model="claude-sonnet-5-5",
    db=db,
    cwd=str(workspace),
    system_prompt="Use the support MCP tools to answer ticket questions. Only save a note when explicitly requested.",
    tools=[],
    allowed_tools=["mcp__support__get_ticket", "mcp__support__get_order", "mcp__support__save_note"],
    permission_mode="dontAsk",
    setting_sources=[],
    strict_mcp_config=True,
    mcp_servers={"support": {"type": "http", "url": "http://127.0.0.1:8001/mcp"}},
)

allowed_tools auto-approves those MCP tools. It is not an exclusive allowlist by itself; this example also disables built-in tools and limits loaded configuration.

Codex

agent = CodexAgent(
    id="support-agent",
    model="gpt-5.6-luna",
    db=db,
    cwd=str(workspace),
    instructions="Use the support MCP tools to answer ticket questions. Only save a note when explicitly requested.",
    sandbox="read-only",
    approval_mode="deny_all",
    mcp_servers={
        "support": {
            "url": "http://127.0.0.1:8001/mcp",
            "required": True,
            "tools": {
                "get_ticket": {"approval_mode": "approve"},
                "get_order": {"approval_mode": "approve"},
                "save_note": {"approval_mode": "approve"},
            },
        }
    },
)

These explicit approvals authorize the three fixture tools, including the write operation. A read-only local filesystem policy does not prevent remote MCP tools from changing their own data.

Verify a real tool result

Restart harness_service.py, then submit:

curl -N http://127.0.0.1:7777/agents/support-agent/runs \
  -F 'message=Use the support MCP tools to read ticket T-100 and its order. Tell me when it will arrive. Do not save a note.' \
  -F 'session_id=support-tools-demo' \
  -F 'stream=true'

Expect calls to get_ticket and get_order, with order O-100 due tomorrow. Fetch the saved run as in the first guide and inspect tools, arguments and actual results. An answer alone does not establish that the tool ran. Inspect tool result content as well as error flags; native error mapping differs between adapters.

Make a controlled change

After the first turn finishes:

curl -N http://127.0.0.1:7777/agents/support-agent/runs \
  -F 'message=Call mcp__support__save_note exactly once with operation_id=note-100, ticket_id=T-100, note=Customer informed of delivery tomorrow. Return the saved note.' \
  -F 'session_id=support-tools-demo' \
  -F 'stream=true'

Verify the receipt independently:

python - <<'PYTHON'
import sqlite3
with sqlite3.connect("data/support-tools.db") as db:
    print(db.execute("SELECT * FROM notes WHERE operation_id=?", ("note-100",)).fetchall())
PYTHON

Repeating the exact operation and arguments must leave one row. A retry that changes arguments receives an explicit conflict. In production, your application should create and preserve operation IDs; a prompt alone is not an idempotency guarantee.

Stop the MCP service with Ctrl-C when finished. See Background Runs and Streaming to run the same task independently of the client connection.