MCP with Open-Weight Models via Tool-Calling Harnesses

Vendor agent SDKs (Claude Agent SDK, OpenAI Agents SDK) bridge a model’s native function-calling to MCP automatically. Self-hosted open-weight models served through a raw inference endpoint (vLLM, SGLang, TGI) need that bridging built explicitly, this is the role of a tool-calling harness. This lesson extends the basic harness to handle parallel tool calls correctly, and closes with a concrete methodology for evaluating tool-calling reliability across models before committing to one in production.

Why a Harness Is Needed

An OpenAI-compatible inference server exposes chat completions with an OpenAI-style tools parameter, but has no concept of MCP’s tools/list or tools/call. A harness sits between the two: it discovers the MCP server’s tools, translates their schemas into the OpenAI function-calling format, includes them on every completion request, and when the model returns a function_call, routes it to the actual MCP server and feeds the result back as the next turn’s tool message.

flowchart LR
    MCP["MCP Server\n(tools/list, tools/call)"] -->|"1. discover tools"| Harness["Tool-Calling Harness"]
    Harness -->|"2. translated to OpenAI\nfunction-calling schema"| Model["Open-weight model\n(vLLM endpoint)"]
    Model -->|"3. function_call response(s)"| Harness
    Harness -->|"4. tools/call\n(one per function_call,\nconcurrent where independent)"| MCP
    MCP -->|"5. result(s), correlated\nby tool_call_id"| Harness
    Harness -->|"6. tool results as next turn"| Model

    style MCP fill:#f0fdf9,stroke:#0D9488,color:#0F172A
    style Harness fill:#fff7ed,stroke:#f59e0b,color:#0F172A
    style Model fill:#EEF0F7,stroke:#6366F1,color:#0F172A

Serving an Open-Weight Model

# Serve gpt-oss-120b behind an OpenAI-compatible endpoint with vLLM
vllm serve openai/gpt-oss-120b \
  --host 0.0.0.0 --port 8000 \
  --tool-call-parser hermes \
  --enable-auto-tool-choice

A Harness That Handles Parallel Tool Calls Correctly

from openai import AsyncOpenAI
from mcp.client.session import ClientSession
import asyncio, json

async def mcp_tools_to_openai_schema(session: ClientSession) -> list[dict]:
    mcp_tools = await session.list_tools()
    return [
        {
            "type": "function",
            "function": {
                "name": t.name.replace(".", "_"),  # OpenAI schema disallows dots
                "description": t.description,
                "parameters": t.inputSchema,
            },
        }
        for t in mcp_tools
    ]

async def execute_single_call(mcp_session: ClientSession, call) -> dict:
    """Execute one tool_call and return a message correlated to its own tool_call_id,
    which the model requires to match results back to the calls it made."""
    mcp_tool_name = call.function.name.replace("_", ".", 1)
    result = await mcp_session.call_tool(mcp_tool_name, json.loads(call.function.arguments))
    return {"role": "tool", "tool_call_id": call.id, "content": json.dumps(result)}

async def run_agent_turn(mcp_session: ClientSession, model_client: AsyncOpenAI, model: str, messages: list[dict]):
    tools = await mcp_tools_to_openai_schema(mcp_session)
    response = await model_client.chat.completions.create(model=model, messages=messages, tools=tools)
    msg = response.choices[0].message

    if msg.tool_calls:
        # Parallel tool calls: execute independently-issued calls concurrently,
        # then append EACH result as its own message, tagged by tool_call_id.
        messages.append(msg.model_dump())
        tool_results = await asyncio.gather(
            *(execute_single_call(mcp_session, call) for call in msg.tool_calls)
        )
        messages.extend(tool_results)
        return await run_agent_turn(mcp_session, model_client, model, messages)  # loop until no more tool calls

    return msg.content

# Point the harness at a self-hosted open-weight model instead of a vendor API
model_client = AsyncOpenAI(base_url="http://localhost:8000/v1", api_key="not-needed")
answer = await run_agent_turn(mcp_session, model_client, model="openai/gpt-oss-120b", messages=[...])

The asyncio.gather over execute_single_call is safe specifically because the model issued these calls in parallel in the first place, meaning it does not expect one call’s result before deciding to make another, if a model instead issues one call, waits, then issues a second call depending on the first’s result, that naturally becomes two separate turns of the loop rather than one batch, and the harness handles both cases correctly without special-casing since the loop structure itself, not a hardcoded parallel/sequential flag, determines it.

The same harness works unchanged against any OpenAI-compatible endpoint, swap model and base_url to point at Kimi K3, GLM-5.2, or any other self-hosted open-weight model with strong agentic tool-calling support, the MCP server and the harness code do not change.

A Concrete Methodology for Evaluating Tool-Calling Reliability

Not every open-weight model produces equally reliable tool calls: well-formed JSON matching the schema, correct argument types, and sound judgment about when a tool call is actually warranted versus answering directly. Before committing to a self-hosted model for a production agent, build a fixed evaluation set and measure three distinct metrics against it, rather than relying on informal impressions from a handful of manual tests.

EVAL_TASKS = [
    {"query": "What's Acme Corp's current subscription tier?", "should_call_tool": True, "expected_tool": "crm.search_accounts"},
    {"query": "What does 'MRR' stand for?", "should_call_tool": False, "expected_tool": None},  # Tests over-calling
    {"query": "Find all high-priority tickets for account A1042 from the last week.", "should_call_tool": True, "expected_tool": "support.search_tickets"},
]

async def evaluate_model(model_client, model: str, mcp_session, tasks: list[dict]) -> dict:
    schema_valid, correct_call_decision, correct_tool_selection = 0, 0, 0
    for task in tasks:
        messages = [{"role": "user", "content": task["query"]}]
        tools = await mcp_tools_to_openai_schema(mcp_session)
        response = await model_client.chat.completions.create(model=model, messages=messages, tools=tools)
        msg = response.choices[0].message

        called_a_tool = bool(msg.tool_calls)
        if called_a_tool == task["should_call_tool"]:
            correct_call_decision += 1  # Metric 2: did it correctly decide whether a tool was needed at all?

        if called_a_tool:
            try:
                json.loads(msg.tool_calls[0].function.arguments)
                schema_valid += 1  # Metric 1: is the JSON well-formed at all?
            except json.JSONDecodeError:
                pass
            if msg.tool_calls[0].function.name.replace("_", ".", 1) == task["expected_tool"]:
                correct_tool_selection += 1  # Metric 3: did it pick the RIGHT tool?

    n = len(tasks)
    return {
        "schema_valid_rate": schema_valid / n,
        "correct_call_decision_rate": correct_call_decision / n,
        "tool_selection_accuracy": correct_tool_selection / n,
    }

The second metric, correct call decision rate, matters as much as the more obvious schema-validity check: a model that over-calls tools for questions answerable directly (like “what does MRR stand for”) wastes latency and cost on unnecessary round-trips, while a model that under-calls tools misses opportunities to ground its answer in real data. Running this evaluation across candidate models with the exact same MCP server and task set gives a defensible, quantitative basis for a model choice, rather than a decision made on vibes from a few manual tries, and it is worth re-running periodically as models are updated, since a provider’s newer checkpoint of the “same” model can shift these numbers meaningfully.

The next lesson scales this pattern up: an agent that needs to call tools across many MCP servers in a single reasoning session, not just one.