Build an MCP Client in Python: Connect Servers to an LLM

# client_basic.py - mcp 2.3.0
import asyncio, sys
from mcp import Client, StdioServerParameters

async def main():
    server = StdioServerParameters(command=sys.executable, args=["batch_server.py"])
    async with Client(server) as client:                 # or Client("http://127.0.0.1:8000/mcp")
        print("Protocol:", client.protocol_version)      # 2026-07-28 against a current server
        print("Server:", client.server_info.name, "-", client.instructions)

        for tool in (await client.list_tools()).tools:
            print("Tool:", tool.name, "| read-only:", bool(tool.annotations and tool.annotations.read_only_hint))

        result = await client.call_tool("find_batch", {"code": "GENAI-EVE"})
        if result.is_error:
            print("Tool failed:", result.content[0].text)
        else:
            print("Structured:", result.structured_content)

asyncio.run(main())

import json

def result_to_model_text(result, max_chars=600):
    parts = []
    for block in result.get("content", []):
        if block["type"] == "text":
            parts.append(block["text"])
        elif block["type"] == "resource_link":
            parts.append(f"[link: {block['name']} -> {block['uri']}]")
        else:
            parts.append(f"[{block['type']} content, {block.get('mimeType', 'unknown type')}]")
    if not parts and "structuredContent" in result:          # very old or minimal servers
        parts.append(json.dumps(result["structuredContent"]))
    text = "\n".join(parts)
    prefix = "TOOL ERROR: " if result.get("isError") else ""
    return prefix + (text if len(text) <= max_chars else text[:max_chars] + " ...[truncated]")

samples = [
    {"content": [{"type": "text", "text": '{"code": "GENAI-EVE", "seats": 3}'}], "structuredContent": {"seats": 3}},
    {"content": [{"type": "text", "text": "Unknown batch 'ROBO'."}], "isError": True},
    {"content": [{"type": "image", "mimeType": "image/png", "data": "iVBORw0..."},
                 {"type": "resource_link", "name": "syllabus.md", "uri": "batches://DS-MOR/syllabus"}]},
    {"content": [{"type": "text", "text": "x" * 900}]},
]
for s in samples:
    out = result_to_model_text(s)
    print(repr(out) if len(out) < 100 else f"{len(out)} chars, ends with {out[-20:]!r}")

servers = {
    "batches": [{"name": "search", "description": "Search batches by topic"},
                {"name": "join_waitlist", "description": "Waitlist a student"}],
    "library": [{"name": "search", "description": "Search books by title"},
                {"name": "check_book", "description": "Copies of a book on the shelf"}],
}

def merge(servers, sep="__"):
    routes, model_tools = {}, []
    for server, tools in servers.items():
        for tool in tools:
            public = f"{server}{sep}{tool['name']}"
            if public in routes:
                raise ValueError(f"Duplicate tool name after prefixing: {public}")
            routes[public] = (server, tool["name"])
            model_tools.append({"type": "function", "name": public, "description": tool["description"]})
    return routes, model_tools

routes, model_tools = merge(servers)
print("Model sees:", [t["name"] for t in model_tools])
print("Route for library__search:", routes["library__search"])
print("Route for batches__search:", routes["batches__search"])

import sys
from contextlib import AsyncExitStack
from mcp import Client, StdioServerParameters

SERVERS = {
    "batches": StdioServerParameters(command=sys.executable, args=["batch_server.py"]),
    "library": StdioServerParameters(command=sys.executable, args=["library_server.py"]),
}

async def connect_all(stack: AsyncExitStack):
    clients, routes, tools, read_only = {}, {}, [], set()
    for server_name, params in SERVERS.items():
        client = clients[server_name] = await stack.enter_async_context(Client(params))
        for tool in (await client.list_tools()).tools:
            public = f"{server_name}__{tool.name}"
            routes[public] = (server_name, tool.name)
            tools.append({"type": "function", "name": public,
                          "description": tool.description or "", "parameters": tool.input_schema})
            if tool.annotations and tool.annotations.read_only_hint:
                read_only.add(public)            # trust hints only from servers you trust
    return clients, routes, tools, read_only

# agent.py - an MCP host loop: mcp 2.3.0 + OpenAI Responses API (pip install openai)
import asyncio, json
from contextlib import AsyncExitStack
from openai import AsyncOpenAI
# connect_all() and SERVERS from the previous block go here

async def run_agent(question: str, max_steps: int = 6) -> str:
    llm = AsyncOpenAI()
    async with AsyncExitStack() as stack:
        clients, routes, tools, read_only = await connect_all(stack)
        items = [{"role": "user", "content": question}]
        for _ in range(max_steps):
            response = await llm.responses.create(model="your-model-name", tools=tools, input=items)
            items += response.output
            calls = [item for item in response.output if item.type == "function_call"]
            if not calls:
                return response.output_text
            for call in calls:
                server_name, tool_name = routes[call.name]
                prompt = f"Allow {call.name}({call.arguments})? [y/N] "
                if call.name not in read_only and (await asyncio.to_thread(input, prompt)).lower() != "y":
                    output = "The user declined this action."
                else:
                    result = await clients[server_name].call_tool(tool_name, json.loads(call.arguments))
                    output = "\n".join(b.text for b in result.content if b.type == "text")
                items.append({"type": "function_call_output", "call_id": call.call_id, "output": output})
        return "Stopped: too many tool steps."

if __name__ == "__main__":
    print(asyncio.run(run_agent("Is the Data Science morning batch full? If so, waitlist asha@example.com.")))

import json

def fake_model(items):
    done = {i["call_id"] for i in items if i.get("type") == "function_call_output"}
    if "c1" not in done:
        return [{"type": "function_call", "call_id": "c1", "name": "batches__find_batch", "arguments": '{"code": "DS-MOR"}'}]
    if "c2" not in done:
        return [{"type": "function_call", "call_id": "c2", "name": "batches__join_waitlist",
                 "arguments": '{"code": "DS-MOR", "student_email": "asha@example.com"}'}]
    return [{"type": "message", "text": "DS-MOR is full. I did not add you to the waitlist, as you asked me not to."}]

def mcp_call(name, args):
    return {"batches__find_batch": '{"code": "DS-MOR", "seats": 0}'}.get(name, "added at position 1")

READ_ONLY, user_says = {"batches__find_batch"}, {"batches__join_waitlist": "n"}
items = [{"role": "user", "content": "Is DS-MOR full? Waitlist me if so."}]
for step in range(6):
    output = fake_model(items)
    items += output
    calls = [o for o in output if o["type"] == "function_call"]
    if not calls:
        print("Answer:", output[0]["text"]); break
    for c in calls:
        allowed = c["name"] in READ_ONLY or user_says.get(c["name"]) == "y"
        result = mcp_call(c["name"], json.loads(c["arguments"])) if allowed else "The user declined this action."
        print(f"step {step}: {c['name']} allowed={allowed} -> {result}")
        items.append({"type": "function_call_output", "call_id": c["call_id"], "output": result})

from mcp import Client
from mcp.types import ElicitResult

async def ask_user(context, params):
    print("Server asks:", params.message)                 # show params.requested_schema as a form
    answer = input("Confirm? [y/N] ").lower() == "y"
    return ElicitResult(action="accept", content={"ok": answer})

# async with Client(server, elicitation_callback=ask_user) as client: ...