AI Agent Tool Loop (MCP Client)
Build the other side of MCP: the client loop that hands tools to a language model, runs the calls it asks for, feeds the results back and stops safely. This is the core of every AI agent.
How it works in the real world
An AI agent is a loop, not magic. The client (Claude Desktop, Claude Code, your own script) does this until the model is finished:
- Offer tools: fetch them from MCP servers with
tools/listand pass them to the model in its tool format. - Ask the model: it replies with text, or with
tool_useblocks naming a tool and its input. - Run the calls: execute each one (through MCP's
tools/call) and send backtool_resultblocks with the same ids. - Repeat or stop: when the model's
stop_reasonis no longertool_use, its text is the answer. A good client also enforces limits so a confused model cannot loop forever.
Here the model is scripted, so every run is identical and gradable. The take-it-further section swaps in the real Claude API.
Translate MCP tools for the model
MCP describes tools with inputSchema; the Claude Messages API expects input_schema. Convert each MCP tool into {"name", "description", "input_schema"} (use an empty string when a description is missing) and print each as JSON.
A list comprehension works: {"name": t["name"], "description": t.get("description", ""), "input_schema": t["inputSchema"]}.
import json
mcp_tools = [
{"name": "get_weather", "description": "Current weather for a city.",
"inputSchema": {"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]}},
{"name": "convert_temp",
"inputSchema": {"type": "object", "properties": {"celsius": {"type": "number"}, "unit": {"type": "string"}}, "required": ["celsius"]}},
]
model_tools = [
{"name": t["name"], "description": t.get("description", ""), "input_schema": t["inputSchema"]}
for t in mcp_tools
]
for t in model_tools:
print(json.dumps(t))
Run your code to check this step…
Route the model's tool calls
The model answered with a mix of text and tool_use blocks. Run every tool call against LOCAL_TOOLS and build a tool_result block for each, carrying the same tool_use_id. Print each result as JSON.
Skip blocks whose type is not tool_use. The result is {"type": "tool_result", "tool_use_id": block["id"], "content": str(value)}.
import json
LOCAL_TOOLS = {
"get_weather": lambda city: {"Vilnius": "18C, clear", "Lisbon": "24C, sunny"}.get(city, "unknown city"),
"convert_temp": lambda celsius, unit="F": round(celsius * 9 / 5 + 32, 1) if unit == "F" else round(celsius + 273.15, 2),
}
response_content = [
{"type": "text", "text": "Let me check both cities."},
{"type": "tool_use", "id": "toolu_01", "name": "get_weather", "input": {"city": "Vilnius"}},
{"type": "tool_use", "id": "toolu_02", "name": "get_weather", "input": {"city": "Lisbon"}},
]
for block in response_content:
if block["type"] != "tool_use":
continue
value = LOCAL_TOOLS[block["name"]](**block["input"])
print(json.dumps({"type": "tool_result", "tool_use_id": block["id"], "content": str(value)}))
Run your code to check this step…
The agent loop
Wire it together. fake_model(messages) plays the model. Loop: call it, append its reply as an assistant message, and if stop_reason is tool_use run the tools and append a user message of tool results; otherwise print the final text. Log each call as -> name(input) and each result as <- value.
while True: reply = fake_model(messages); messages.append({"role": "assistant", "content": reply["content"]}); if reply["stop_reason"] != "tool_use": print the text and break. json.dumps(block["input"]) formats the logged input.
import json
LOCAL_TOOLS = {
"get_weather": lambda city: {"Vilnius": "18C, clear", "Lisbon": "24C, sunny"}.get(city, "unknown city"),
"convert_temp": lambda celsius, unit="F": round(celsius * 9 / 5 + 32, 1) if unit == "F" else round(celsius + 273.15, 2),
}
SCRIPT = [
{"stop_reason": "tool_use", "content": [
{"type": "tool_use", "id": "t1", "name": "get_weather", "input": {"city": "Vilnius"}}]},
{"stop_reason": "tool_use", "content": [
{"type": "tool_use", "id": "t2", "name": "convert_temp", "input": {"celsius": 18}}]},
{"stop_reason": "end_turn", "content": [
{"type": "text", "text": "Vilnius is 18C (64.4F) and clear: a good day for a walk."}]},
]
def fake_model(messages):
turns = sum(1 for m in messages if m["role"] == "assistant")
return SCRIPT[turns]
messages = [{"role": "user", "content": "What is the weather in Vilnius, in Fahrenheit?"}]
while True:
reply = fake_model(messages)
messages.append({"role": "assistant", "content": reply["content"]})
if reply["stop_reason"] != "tool_use":
print("".join(b["text"] for b in reply["content"] if b["type"] == "text"))
break
results = []
for block in reply["content"]:
if block["type"] == "tool_use":
print(f"-> {block['name']}({json.dumps(block['input'])})")
value = LOCAL_TOOLS[block["name"]](**block["input"])
print(f"<- {value}")
results.append({"type": "tool_result", "tool_use_id": block["id"], "content": str(value)})
messages.append({"role": "user", "content": results})
print(f"{len(messages)} messages in the conversation")
Run your code to check this step…
Guardrails
Real agents need limits. Run this misbehaving script with two rules: a tool that is not in ALLOWED returns a tool_result with is_error: True and the text Tool not allowed: NAME (the model sees it and adapts), and after MAX_CALLS tool calls the loop stops with Stopped: tool budget of 3 calls reached.
Check the budget before running each call: if calls == MAX_CALLS, print the stop message and end the loop. Count a call even when it is refused, because refusals still cost a turn.
import json
LOCAL_TOOLS = {
"get_weather": lambda city: {"Vilnius": "18C, clear", "Lisbon": "24C, sunny"}.get(city, "unknown city"),
"delete_files": lambda path: f"deleted {path}",
}
ALLOWED = {"get_weather"}
MAX_CALLS = 3
SCRIPT = [
{"stop_reason": "tool_use", "content": [{"type": "tool_use", "id": "t1", "name": "delete_files", "input": {"path": "/"}}]},
{"stop_reason": "tool_use", "content": [{"type": "tool_use", "id": "t2", "name": "get_weather", "input": {"city": "Lisbon"}}]},
{"stop_reason": "tool_use", "content": [{"type": "tool_use", "id": "t3", "name": "get_weather", "input": {"city": "Lisbon"}}]},
{"stop_reason": "tool_use", "content": [{"type": "tool_use", "id": "t4", "name": "get_weather", "input": {"city": "Lisbon"}}]},
{"stop_reason": "end_turn", "content": [{"type": "text", "text": "never reached"}]},
]
def fake_model(messages):
return SCRIPT[sum(1 for m in messages if m["role"] == "assistant")]
messages = [{"role": "user", "content": "Clean up my disk, then tell me the weather in Lisbon."}]
calls = 0
stopped = False
while not stopped:
reply = fake_model(messages)
messages.append({"role": "assistant", "content": reply["content"]})
if reply["stop_reason"] != "tool_use":
print("".join(b["text"] for b in reply["content"] if b["type"] == "text"))
break
results = []
for block in reply["content"]:
if block["type"] != "tool_use":
continue
if calls == MAX_CALLS:
print(f"Stopped: tool budget of {MAX_CALLS} calls reached")
stopped = True
break
calls += 1
if block["name"] not in ALLOWED:
print(f"x {block['name']} refused")
results.append({"type": "tool_result", "tool_use_id": block["id"],
"content": f"Tool not allowed: {block['name']}", "is_error": True})
continue
value = LOCAL_TOOLS[block["name"]](**block["input"])
print(f"-> {block['name']}({json.dumps(block['input'])}) <- {value}")
results.append({"type": "tool_result", "tool_use_id": block["id"], "content": str(value)})
messages.append({"role": "user", "content": results})
print(f"tool calls used: {calls}")
Run your code to check this step…
Project shipped!
You just built a complete, working ai agent tool loop (mcp client): the same architecture running in production software. Pick your next build →
Take it further
Swap the scripted model for Claude and the local tools for a real MCP server: the loop you just wrote stays almost the same. Nothing below is graded: it runs on your own machine, not in this tab.
Stretch goals
- Connect two MCP servers at once and merge their tool lists, prefixing names so they cannot collide.
- Ask the user for confirmation before any tool whose name starts with
delete_orsend_. - Log every tool call with its duration to a JSON file, then build a report of the slowest tools.
- Stop the loop when the conversation exceeds a token budget, not just a call budget.
agent.py: Claude + your MCP server (needs an Anthropic API key)
# pip install anthropic "mcp[cli]"
# export ANTHROPIC_API_KEY=...
import asyncio
import anthropic
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
async def main():
server = StdioServerParameters(command="python", args=["server.py"])
async with stdio_client(server) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
listed = await session.list_tools()
tools = [{"name": t.name, "description": t.description or "",
"input_schema": t.inputSchema} for t in listed.tools]
client = anthropic.Anthropic()
messages = [{"role": "user", "content": "What is 19 + 23, and what is 21.5C in F?"}]
for _ in range(10): # the guardrail from step 4
reply = client.messages.create(model="claude-sonnet-5", max_tokens=1024,
tools=tools, messages=messages)
messages.append({"role": "assistant", "content": reply.content})
if reply.stop_reason != "tool_use":
print("".join(b.text for b in reply.content if b.type == "text"))
break
results = []
for block in reply.content:
if block.type == "tool_use":
out = await session.call_tool(block.name, block.input)
text = "".join(c.text for c in out.content if c.type == "text")
results.append({"type": "tool_result", "tool_use_id": block.id,
"content": text, "is_error": out.isError})
messages.append({"role": "user", "content": results})
asyncio.run(main())
Ask an AI to review your build
I wrote an AI agent loop in Python: it converts MCP tools to the Claude tool format, runs tool_use blocks, sends tool_result blocks back and enforces an allowlist and a call budget. Here is my code: [paste it]. What failure modes am I not handling (parallel tool calls, tools that hang, the model repeating the same call)? Suggest the single most valuable guardrail to add next and explain why.