Building MCP Clients
from Scratch

MCP Client Architecture

Python Client with fastmcp

from fastmcp import Client
from fastmcp.client.transports import SSETransport, StdioTransport
import asyncio, json

async def run_mcp_workflow():
    # Option 1: Connect to a remote SSE server
    transport = SSETransport(
        "https://api.example.com/mcp",
        headers={"Authorization": "Bearer YOUR_KEY"}
    )

    # Option 2: Connect to a local stdio server
    # transport = StdioTransport("python", ["-m", "my_mcp_server"])

    async with Client(transport) as client:
        # Discover what's available
        tools = await client.list_tools()
        print(f"Available tools: {[t.name for t in tools]}")

        # Call a tool
        result = await client.call_tool(
            "search_documents",
            {"query": "AWS security best practices", "max_results": 5}
        )
        print(f"Results: {result}")

        # Read a resource
        resource = await client.read_resource("schema://database/users")
        print(f"Schema: {resource}")

asyncio.run(run_mcp_workflow())

TypeScript Client SDK

import { Client } from "@modelcontextprotocol/sdk/client/index.js"
import { SSEClientTransport } from "@modelcontextprotocol/sdk/client/sse.js"

async function mcpWorkflow() {
  const transport = new SSEClientTransport(
    new URL("https://api.example.com/mcp"),
    { headers: { Authorization: `Bearer ${process.env.MCP_KEY}` } }
  )
  const client = new Client(
    { name: "my-client", version: "1.0.0" },
    { capabilities: {} }
  )
  await client.connect(transport)

  // List tools
  const { tools } = await client.listTools()
  console.log("Tools:", tools.map(t => t.name))

  // Call a tool
  const result = await client.callTool({
    name: "search_documents",
    arguments: { query: "IAM policies", maxResults: 5 }
  })
  console.log("Result:", result.content[0].text)

  await client.close()
}
mcpWorkflow().catch(console.error)

Agentic Client — Auto Tool Selection

from fastmcp import Client
import anthropic, json

async def agentic_mcp_client(goal: str, mcp_server_url: str):
    claude = anthropic.Anthropic()

    async with Client(SSETransport(mcp_server_url)) as mcp:
        # Convert MCP tools to Anthropic tool format
        mcp_tools = await mcp.list_tools()
        anthropic_tools = [{
            "name": t.name,
            "description": t.description,
            "input_schema": t.inputSchema
        } for t in mcp_tools]

        messages = [{"role": "user", "content": goal}]

        while True:
            response = claude.messages.create(
                model="claude-sonnet-4-5", max_tokens=4096,
                tools=anthropic_tools, messages=messages
            )
            if response.stop_reason == "end_turn":
                return response.content[0].text  # Done!

            # Execute tool calls
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    result = await mcp.call_tool(block.name, block.input)
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": str(result)
                    })
            messages.append({"role": "assistant", "content": response.content})
            messages.append({"role": "user", "content": tool_results})

Web Dashboard MCP Client (React)

// Next.js API route — acts as MCP proxy for the React frontend
// pages/api/mcp-tool.ts
import { Client } from "@modelcontextprotocol/sdk/client/index.js"
import { SSEClientTransport } from "@modelcontextprotocol/sdk/client/sse.js"
import type { NextApiRequest, NextApiResponse } from "next"

// Reuse connection across requests (connection pooling)
let _client: Client | null = null
async function getClient() {
  if (!_client) {
    _client = new Client({ name: "dashboard", version: "1.0" }, { capabilities: {} })
    const transport = new SSEClientTransport(new URL(process.env.MCP_SERVER_URL!))
    await _client.connect(transport)
  }
  return _client
}

export default async function handler(req: NextApiRequest, res: NextApiResponse) {
  const { toolName, args } = req.body
  const client = await getClient()
  const result = await client.callTool({ name: toolName, arguments: args })
  res.json({ result: result.content[0].text })
}

CLI Tool Client Pattern

#!/usr/bin/env python3
# mcp-cli.py — invoke any MCP tool from the command line
# Usage: python mcp-cli.py search_documents '{"query": "AWS IAM", "max_results": 3}'
import sys, asyncio, json
from fastmcp import Client
from fastmcp.client.transports import SSETransport

async def main():
    tool_name = sys.argv[1]
    tool_args = json.loads(sys.argv[2]) if len(sys.argv) > 2 else {}

    server_url = os.environ.get("MCP_SERVER", "http://localhost:8080")
    async with Client(SSETransport(server_url)) as client:
        result = await client.call_tool(tool_name, tool_args)
        print(json.dumps(result, indent=2))

asyncio.run(main())