← Kembali ke Blog

Bikin MCP Client yang Nyambung ke Banyak Server Sekaligus (Python)

Kebanyakan tutorial MCP yang aku baca cuma nyambungin satu client ke satu server, list tool-nya, terus udah. Itu cukup buat demo. Tapi ambruk begitu kamu mau bikin agen yang bisa baca file, query Postgres, cari di web, dan post ke Slack semuanya dalam satu turn. Python SDK-nya emang didesain buat skenario ini. Tutorial-nya aja yang jarang nunjukin.

Artikel ini bikin client kecil yang nyambung ke dua MCP server sekaligus: math server (beberapa tool aritmatika) dan notes server (tool buat nyimpen dan ambil string pendek). Dia ngambil gabungan daftar tool-nya, tulis ulang namanya pakai prefix server biar nggak bisa tabrakan, dan oper hasilnya ke Claude sebagai tool definition format Anthropic. Pas model manggil tool, client cari tau server mana yang punya, dispatch panggilan-nya, dan masukin hasilnya balik ke percakapan.

Hasilnya agen multi-server yang jalan dalam sekitar 150 baris Python. Nggak ada framework di atasnya, nggak ada LangChain, nggak ada abstraksi yang harus dipelajarin sebelum kamu bisa debug.

Prerequisites

  • Python 3.10 atau lebih baru
  • uv buat dependency management (pip install uv kalau belum punya)
  • Anthropic API key di shell sebagai ANTHROPIC_API_KEY
  • Sekitar 5 menit

MCP Python SDK ada di versi 1.28.1 waktu artikel ini ditulis. Pin ke mcp>=1.27,<2 kalau kamu deploy ini di mana pun; v2 masih alpha (pre-release sekarang 2.0.0a3) dan import path sama signature transport-nya beda.

Step 1: Bikin Dua Server Kecil

Kamu butuh sesuatu buat dikonekin. Dua server minimal lebih gampang dipikirin daripada satu yang gede, dan masalah tabrakan nama cuma muncul kalau kamu punya lebih dari satu.

mkdir multi-mcp && cd multi-mcp
uv init --no-readme .
uv add "mcp[cli]" anthropic

Ganti isi math_server.py:

"""Server MCP math yang minimal."""
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("math")


@mcp.tool()
def add(a: float, b: float) -> float:
    """Tambah dua angka dan balikin hasilnya."""
    return a + b


@mcp.tool()
def multiply(a: float, b: float) -> float:
    """Kaliin dua angka dan balikin hasilnya."""
    return a * b


if __name__ == "__main__":
    mcp.run(transport="stdio")

Dan notes_server.py:

"""Server MCP notes yang minimal."""
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("notes")
_STORE: dict[str, str] = {}


@mcp.tool()
def save(key: str, value: str) -> str:
    """Simpen string di bawah sebuah key. Timpa kalau key-nya udah ada."""
    _STORE[key] = value
    return f"saved {len(value)} chars under '{key}'"


@mcp.tool()
def recall(key: str) -> str:
    """Ambil string berdasarkan key. Balikin 'not found' kalau key-nya nggak ada."""
    return _STORE.get(key, "not found")


if __name__ == "__main__":
    mcp.run(transport="stdio")

Dua-duanya server MCP yang valid. Mereka ngomong protokol yang sama kayak server resmi, dan itu inti dari standard-nya: siapa pun bisa nulis server baru dan client mana pun bisa pake.

Step 2: Nyambung ke Dua Server dalam Satu Client

SDK-nya ngasih kamu stdio_client buat transport subprocess dan ClientSession buat protokolnya sendiri. Nyambung ke N server tinggal bungkus N pasang dari itu pake asyncio.gather. Object session-nya independen, dan itu fine. Mereka emang nggak share apa-apa secara desain.

Bikin client.py:

"""Nyambung ke dua MCP server dan expose tool-nya sebagai tool definition Anthropic."""
import asyncio
from contextlib import asynccontextmanager
from typing import AsyncIterator

from anthropic import Anthropic
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

SERVERS = {
    "math": StdioServerParameters(command="python", args=["math_server.py"]),
    "notes": StdioServerParameters(command="python", args=["notes_server.py"]),
}


@asynccontextmanager
async def multi_session() -> AsyncIterator[dict[str, ClientSession]]:
    """Buka satu ClientSession per server, paralel, dan yield berdasarkan nama."""
    sessions: dict[str, ClientSession] = {}

    async def _open(name: str, params: StdioServerParameters) -> None:
        # Tiap stdio_client() launch subprocess-nya buat kita.
        async with stdio_client(params) as (read, write):
            session = ClientSession(read, write)
            await session.initialize()
            sessions[name] = session

    # Buka semua server secara concurrent. Kalau satu gagal, yang lain masih clean up.
    try:
        await asyncio.gather(*(_open(n, p) for n, p in SERVERS.items()))
        yield sessions
    finally:
        # ClientSession nggak punya close eksplisit di 1.x; context manager di atas
        # yang handle teardown subprocess pas fungsi balikin.
        pass


async def list_all_tools(sessions: dict[str, ClientSession]) -> list[dict]:
    """Balikin tool definition format Anthropic buat semua tool dari semua server.

    Nama ditulis ulang jadi '<server>__<tool>' biar tool yang namanya 'delete' di
    dua server nggak bisa tabrakan pas Claude liat.
    """
    tools: list[dict] = []
    for server_name, session in sessions.items():
        result = await session.list_tools()
        for tool in result.tools:
            tools.append({
                "name": f"{server_name}__{tool.name}",
                "description": tool.description or "",
                "input_schema": tool.inputSchema,
            })
    return tools


async def dispatch_tool_call(
    sessions: dict[str, ClientSession], name: str, arguments: dict
) -> str:
    """Rutein tool_use block dari Claude ke server yang tepat dan balikin output text-nya."""
    server_name, _, tool_name = name.partition("__")
    session = sessions[server_name]
    result = await session.call_tool(tool_name, arguments=arguments)
    # Tiap content block punya atribut .text. Gabungin yang tipenya text.
    parts = []
    for block in result.content:
        text = getattr(block, "text", None)
        if text is not None:
            parts.append(text)
    return "
".join(parts) if parts else "(no output)"

Separator __ itu konvensi yang sama kayak yang dipake Claude Agent SDK di internal. Pilih ini dari awal artinya tool yang kamu tulis nanti bisa pindah antara standalone client sama SDK tanpa rename.

Step 3: Drive Tool Loop-nya Pake Claude

Ini bagian yang SDK-nya nggak lakuin buat kamu, karena ngerouting loop Anthropic Messages API itu tugas host. Sekitar 40 baris.

async def run_conversation(prompt: str) -> str:
    anthropic = Anthropic()
    async with multi_session() as sessions:
        tools = await list_all_tools(sessions)
        messages = [{"role": "user", "content": prompt}]

        # Loop dengan cap keras. Model bisa spiral, dan kamu mau sadar pas itu kejadian.
        for _ in range(8):
            response = anthropic.messages.create(
                model="claude-sonnet-4-5",
                max_tokens=1024,
                tools=tools,
                messages=messages,
            )

            # Kalau model nggak manggil tool, kita udah selesai.
            if response.stop_reason != "tool_use":
                return "".join(
                    b.text for b in response.content if hasattr(b, "text")
                )

            # Append turn assistant, terus jalanin semua tool call yang dia minta.
            messages.append({"role": "assistant", "content": response.content})
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    output = await dispatch_tool_call(
                        sessions, block.name, block.input
                    )
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": output,
                    })
            messages.append({"role": "user", "content": tool_results})

        return "(gave up after 8 turns)"


if __name__ == "__main__":
    question = "Multiply 17 by 23, then save the answer under the key 'magic'."
    print(asyncio.run(run_conversation(question)))

Dua hal yang perlu diperhatiin. Pertama, model bisa ngeluarin lebih dari satu tool call dalam satu turn. Kalau Claude mutusin panggil math__add sama notes__save di response yang sama, loop di atas handle dua-duanya sebelum model call berikutnya. Kedua, tool_use_id ngeikat tiap result balik ke tool_use block yang bener. Skip ini adalah sumber paling umum dari error "model komplain soal tool ID yang nggak cocok".

Step 4: Jalanin

export ANTHROPIC_API_KEY=sk-ant-...
uv run python client.py

Kamu bakal liat sesuatu kayak 391 diikuti baris konfirmasi. Model manggil math__multiply, dapet 391, terus manggil notes__save dengan {"key": "magic", "value": "391"}. Kalau kamu jalanin lagi dan tanya "what is stored under 'magic'?", dia bakal manggil notes__recall dan jawab 391. Dua server, satu percakapan, nggak ada lem custom.

Jebakan yang Sering Kejadian

Lupa konvensi separator. Dua tool yang dua-duanya namanya search di server beda bakal di-overwrite diem-diem di daftar tool Claude. Selalu prefix sama nama server. Format server_name__tool_name itu standar de-facto, dan Claude Agent SDK pake format yang sama.

Baca tool.inputSchema bukan input_schema. Response server MCP pake camelCase (inputSchema). Anthropic Messages API pake snake_case (input_schema). Di contoh di atas dilewatin apa adanya, tapi pas kamu reshape buat alasan apa pun, kamu bakal ke-trip di sini.

Biarin loop jalan terus. Selalu kasih cap turn. Model yang nggak bisa nentuin tool mana yang harus dipanggil bakal terus pilih yang salah. Delapan turn itu default waras buat satu pertanyaan user. Kalau kamu ngerasa butuh lebih, biasanya pertanyaannya terlalu vague, bukan loop-nya yang kebetulan pendek.

Campur tipe transport tanpa mikir. Server lokal lewat stdio itu simpel. Server remote pake streamable HTTP transport dan butuh client sendiri. mcp.client.streamable_http.streamable_http_client itu entry point-nya. Jangan coba bridge di level session; bridge di level dispatcher (satu fungsi per transport, interface yang sama balik ke loop).

Asumsi deskripsi tool itu opsional. Secara teknis emang opsional di schema. Tapi model ngeperlakuinnya sebagai wajib di praktik. Tool dengan deskripsi kosong ya tool yang nggak bakal dipilih model, atau dipilih karena alasan yang salah. Abisin 30 detik buat nulis kalimat beneran.

Kapan Pake Pattern Ini

Pake client multi-server waktu toolset agen-nya nyeberang domain boundary: data, file, komunikasi, eksekusi kode. Protokolnya lagi kerja beneran di sini. Tiap server dimiliki tim yang beda, ditulis di bahasa yang beda, di-deploy dengan schedule yang beda. MCP bikin kamu bisa swap salah satunya tanpa nyentuh yang lain.

Skip buat satu tool. Client 150 baris buat satu server itu lebih banyak kode daripada function call langsung. Break-even point itu sekitar dua server, dan makin banyak yang ditambah makin kebayar.

Kalau kamu mau loop, handling transport, permission model, sama subagent API yang udah diurusin, lihat Claude Agent SDK. Dia ngomong MCP secara native dan bisa deklarasi banyak server di ClaudeAgentOptions. Kamu ngasih up async for loop yang eksplisit, dan kamu dapet balik sekitar 400 baris kode yang nggak harus kamu maintain.

Bacaan Lanjutan

Butuh Bantuan Implementasi?

Saya membantu tim mendesain dan membangun infrastruktur cloud scalable, pipeline DevOps, dan sistem production-grade.

Konsultasi Gratis