Function Calling at Scale with MCP (Model Context Protocol): Complete Guide

Function Calling at Scale with MCP (Model Context Protocol): Complete Guide

Function calling has transformed how large language models interact with external systems, but as applications grow, managing dozens of tools across multiple models, sessions, and services becomes a serious engineering challenge. The Model Context Protocol (MCP) — introduced by Anthropic — provides a standardized way to expose tools, resources, and prompts to LLMs. This guide walks through everything you need to build, deploy, and scale function calling systems using MCP.

What Is the Model Context Protocol?

MCP is an open protocol that standardizes communication between AI applications (clients) and external data sources or tools (servers). Think of it as USB-C for AI: a universal interface that lets any MCP-compatible client — Claude Desktop, IDE extensions, custom agents — connect to any MCP server without bespoke integration code.

An MCP server can expose three primitives:

• Tools — Functions the model can invoke (e.g., query_database, send_email).
• Resources — Read-only data the model can access (e.g., files, API responses).
• Prompts — Reusable prompt templates the client can inject.

Function calling at scale means going beyond a handful of hardcoded tools in a single script. It involves managing tool registries, handling concurrent invocations, enforcing security boundaries, routing requests intelligently, and observing behavior across many sessions.

Why MCP Matters for Function Calling at Scale

Before MCP, every framework had its own way of defining tools. OpenAI function calling, LangChain tools, and custom integrations all used different schemas. This created lock-in and made it hard to swap models or share tooling across teams. MCP solves several scaling problems:

• Standardization — One schema (JSON Schema) and one transport protocol for all tools.
• Decoupling — Tool servers run as separate processes or services, independently deployable.
• Composability — A single client can connect to multiple MCP servers simultaneously.
• Language agnosticism — Servers can be written in Python, TypeScript, Go, or anything that speaks the protocol.
• Security isolation — Tools run in their own process with their own permissions.

Architecture Overview

A typical scaled MCP deployment has the following components:

• MCP Clients — Applications that host the LLM and initiate tool calls (e.g., an agent runtime, a chat backend).
• MCP Servers — Processes that expose grouped tools (e.g., a "database server", a "GitHub server", a "search server").
• Transport Layer — stdio for local servers, HTTP/SSE or WebSocket for remote servers.
• Tool Registry — A catalog that aggregates tool definitions from all connected servers.
• Orchestrator — Logic that decides which tools to present to the model and how to route invocations.

Local vs. Remote Transports

For development, stdio transport is simplest: the client spawns the server as a subprocess and communicates over stdin/stdout. For production at scale, you will want HTTP-based transports so servers can run as independent microservices behind load balancers.

Setting Up an MCP Server

The official MCP SDKs are available for TypeScript and Python. Let's build a Python MCP server that exposes a set of tools for a hypothetical e-commerce analytics platform.

First, install the SDK:

pip install mcp

Now create the server:

# server.py
from mcp.server.fastmcp import FastMCP
import asyncio
import json

mcp = FastMCP("ecommerce-analytics")

# Simulated data store
ORDERS = [
    {"id": "ORD-001", "customer": "alice@example.com", "total": 129.99, "status": "shipped"},
    {"id": "ORD-002", "customer": "bob@example.com", "total": 45.50, "status": "pending"},
    {"id": "ORD-003", "customer": "alice@example.com", "total": 210.00, "status": "shipped"},
]

@mcp.tool()
def get_orders(customer_email: str, limit: int = 10) -> str:
    """Retrieve recent orders for a customer.

    Args:
        customer_email: The customer's email address.
        limit: Maximum number of orders to return (default 10).
    """
    results = [o for o in ORDERS if o["customer"] == customer_email][:limit]
    return json.dumps(results)

@mcp.tool()
def calculate_revenue(start_date: str, end_date: str) -> str:
    """Calculate total revenue between two dates.

    Args:
        start_date: ISO date string (e.g., 2024-01-01).
        end_date: ISO date string (e.g., 2024-12-31).
    """
    total = sum(o["total"] for o in ORDERS if o["status"] == "shipped")
    return json.dumps({"start_date": start_date, "end_date": end_date, "revenue": total})

@mcp.tool()
def update_order_status(order_id: str, new_status: str) -> str:
    """Update the status of an existing order.

    Args:
        order_id: The unique order identifier.
        new_status: One of 'pending', 'shipped', 'delivered', 'cancelled'.
    """
    valid = {"pending", "shipped", "delivered", "cancelled"}
    if new_status not in valid:
        return json.dumps({"error": f"Invalid status. Must be one of {valid}"})
    for order in ORDERS:
        if order["id"] == order_id:
            order["status"] = new_status
            return json.dumps({"success": True, "order": order})
    return json.dumps({"error": f"Order {order_id} not found"})

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

The FastMCP class uses decorators to register tools. The docstring and type hints are automatically converted into the JSON Schema that the LLM sees. This is one of MCP's key ergonomics: your function definitions are your tool specifications.

Building the Client

The client connects to one or more MCP servers, collects their tool definitions, and presents them to the LLM. Here is a client that connects to our server and uses Claude for orchestration:

# client.py
import asyncio
import json
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from anthropic import Anthropic

anthropic = Anthropic()

SERVER_PARAMS = StdioServerParameters(
    command="python",
    args=["server.py"],
)

async def run_agent(user_message: str):
    async with stdio_client(SERVER_PARAMS) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # Discover available tools
            tools_response = await session.list_tools()
            tools = []
            for tool in tools_response.tools:
                tools.append({
                    "name": tool.name,
                    "description": tool.description,
                    "input_schema": tool.inputSchema,
                })

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

            # Agentic loop: call model, execute tools, repeat
            while True:
                response = anthropic.messages.create(
                    model="claude-sonnet-4-20250514",
                    max_tokens=4096,
                    tools=tools,
                    messages=messages,
                )

                if response.stop_reason == "end_turn":
                    for block in response.content:
                        if block.type == "text":
                            print(block.text)
                    break

                # Process tool use requests
                assistant_content = response.content
                messages.append({"role": "assistant", "content": assistant_content})

                tool_results = []
                for block in assistant_content:
                    if block.type == "tool_use":
                        print(f"Calling tool: {block.name} with args: {block.input}")
                        result = await session.call_tool(block.name, block.input)
                        tool_results.append({
                            "type": "tool_result",
                            "tool_use_id": block.id,
                            "content": result.content[0].text if result.content else "",
                        })

                messages.append({"role": "user", "content": tool_results})

if __name__ == "__main__":
    asyncio.run(run_agent("What is the total revenue for Alice's shipped orders?"))

This implements the core agentic loop: the model proposes tool calls, the client executes them via the MCP session, and results are fed back until the model produces a final answer.

Connecting to Multiple Servers

At scale, you will have many MCP servers — one per domain or service. The client should connect to all of them and merge their tool catalogs. Here is a multi-server manager:

# multi_client.py
import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

class MCPManager:
    def __init__(self):
        self.sessions = {}  # name -> (session, cleanup)

    async def connect(self, name: str, command: str, args: list[str]):
        params = StdioServerParameters(command=command, args=args)
        # Keep the context managers alive
        ctx = stdio_client(params)
        read, write = await ctx.__aenter__()
        session = ClientSession(read, write)
        await session.__aenter__()
        await session.initialize()
        self.sessions[name] = (session, ctx)

    async def list_all_tools(self) -> list[dict]:
        all_tools = []
        for server_name, (session, _) in self.sessions.items():
            resp = await session.list_tools()
            for tool in resp.tools:
                all_tools.append({
                    "name": f"{server_name}__{tool.name}",
                    "description": f"[{server_name}] {tool.description}",
                    "input_schema": tool.inputSchema,
                    "_server": server_name,
                    "_original_name": tool.name,
                })
        return all_tools

    async def call_tool(self, namespaced_name: str, arguments: dict):
        # Parse the namespace
        server_name, original_name = namespaced_name.split("__", 1)
        session, _ = self.sessions[server_name]
        return await session.call_tool(original_name, arguments)

    async def close(self):
        for session, ctx in self.sessions.values():
            await session.__aexit__(None, None, None)
            await ctx.__aexit__(None, None, None)

Namespacing tool names with the server prefix prevents collisions when two servers expose tools with the same name (e.g., both a "search" tool from a GitHub server and a "search" tool from a docs server).

Scaling Strategies

1. Tool Filtering and Dynamic Selection

When you have 50+ tools, sending all of them in every request wastes tokens and confuses the model. Implement a retrieval step that selects the most relevant tools based on the user query:

import openai

def select_relevant_tools(query: str, all_tools: list[dict], top_k: int = 10) -> list[dict]:
    """Use embeddings to select the most relevant tools for a query."""
    tool_descriptions = [f"{t['name']}: {t['description']}" for t in all_tools]
    
    response = openai.embeddings.create(
        model="text-embedding-3-small",
        input=[query] + tool_descriptions,
    )
    
    query_embedding = response.data[0].embedding
    tool_embeddings = [item.embedding for item in response.data[1:]]
    
    # Cosine similarity
    scores = []
    for i, tool_emb in enumerate(tool_embeddings):
        dot = sum(a * b for a, b in zip(query_embedding, tool_emb))
        scores.append((dot, i))
    
    scores.sort(reverse=True)
    return [all_tools[idx] for _, idx in scores[:top_k]]

This reduces the tool list from potentially hundreds to a focused subset, dramatically improving both cost and accuracy.

2. Connection Pooling for Remote Servers

When using HTTP transport, maintain a pool of persistent connections rather than creating new ones per request. This reduces latency from TLS handshakes:

import aiohttp
from mcp.client.sse import sse_client

class ConnectionPool:
    def __init__(self, max_connections: int = 20):
        self.max_connections = max_connections
        self.pool: dict[str, list] = {}
        self.semaphore = asyncio.Semaphore(max_connections)

    async def get_session(self, server_url: str):
        async with self.semaphore:
            if server_url not in self.pool:
                self.pool[server_url] = []
            if self.pool[server_url]:
                return self.pool[server_url].pop()
            # Create new connection
            ctx = sse_client(server_url)
            read, write = await ctx.__aenter__()
            session = ClientSession(read, write)
            await session.__aenter__()
            await session.initialize()
            return session, ctx, server_url

    async def return_session(self, session_info):
        session, ctx, server_url = session_info
        self.pool[server_url].append(session_info)

3. Concurrent Tool Execution

When the model requests multiple independent tool calls in a single response, execute them concurrently:

async def execute_tool_calls(session, tool_calls: list):
    """Execute multiple tool calls concurrently."""
    tasks = []
    for call in tool_calls:
        task = session.call_tool(call.name, call.arguments)
        tasks.append(task)
    results = await asyncio.gather(*tasks, return_exceptions=True)
    
    processed = []
    for call, result in zip(tool_calls, results):
        if isinstance(result, Exception):
            processed.append({
                "tool_use_id": call.id,
                "content": f"Error: {str(result)}",
                "is_error": True,
            })
        else:
            processed.append({
                "tool_use_id": call.id,
                "content": result.content[0].text if result.content else "",
            })
    return processed

4. Caching Tool Results

Many tool calls are idempotent (e.g., reading a file, querying a static dataset). Cache these results to avoid redundant work:

import hashlib
import time

class ToolCache:
    def __init__(self, ttl_seconds: int = 300):
        self.cache: dict[str, tuple] = {}  # key -> (value, timestamp)
        self.ttl = ttl_seconds

    def _key(self, tool_name: str, arguments: dict) -> str:
        raw = f"{tool_name}:{json.dumps(arguments, sort_keys=True)}"
        return hashlib.sha256(raw.encode()).hexdigest()

    def get(self, tool_name: str, arguments: dict):
        key = self._key(tool_name, arguments)
        if key in self.cache:
            value, ts = self.cache[key]
            if time.time() - ts < self.ttl:
                return value
            del self.cache[key]
        return None

    def set(self, tool_name: str, arguments: dict, value: str):
        key = self._key(tool_name, arguments)
        self.cache[key] = (value, time.time())

Security Best Practices

Function calling at scale introduces significant security surface area. Every tool is a potential entry point for the model to affect real-world systems.

Principle of Least Privilege

Each MCP server should have only the permissions it needs. A read-only analytics server should never have write access to the database. Use separate database users with scoped permissions per server.

Human-in-the-Loop for Destructive Operations

DESTRUCTIVE_TOOLS = {"update_order_status", "delete_record", "send_email"}

async def call_tool_with_approval(session, tool_name, arguments, require_approval=True):
    if require_approval and tool_name in DESTRUCTIVE_TOOLS:
        print(f"\n⚠️  Approval required for: {tool_name}")
        print(f"   Arguments: {json.dumps(arguments, indent=2)}")
        approval = input("   Approve? (y/n): ")
        if approval.lower() != "y":
            return {"error": "Tool call rejected by user"}
    
    return await session.call_tool(tool_name, arguments)

Input Validation

Never trust model-generated arguments blindly. Validate them server-side before execution:

from pydantic import BaseModel, EmailStr, field_validator

class GetOrdersInput(BaseModel):
    customer_email: EmailStr
    limit: int = 10

    @field_validator("limit")
    @classmethod
    def validate_limit(cls, v):
        if v < 1 or v > 100:
            raise ValueError("limit must be between 1 and 100")
        return v

@mcp.tool()
def get_orders(customer_email: str, limit: int = 10) -> str:
    try:
        validated = GetOrdersInput(customer_email=customer_email, limit=limit)
    except Exception as e:
        return json.dumps({"error": f"Validation failed: {e}"})
    # ... proceed with validated data

Rate Limiting

Implement rate limiting per session and per tool to prevent runaway agents from overwhelming downstream services:

import time
from collections import defaultdict

class RateLimiter:
    def __init__(self, max_calls: int = 60, window_seconds: int = 60):
        self.max_calls = max_calls
        self.window = window_seconds
        self.calls: dict[str, list[float]] = defaultdict(list)

    def check(self, key: str) -> bool:
        now = time.time()
        self.calls[key] = [t for t in self.calls[key] if now - t < self.window]
        if len(self.calls[key]) >= self.max_calls:
            return False
        self.calls[key].append(now)
        return True

Observability and Monitoring

At scale, you need visibility into which tools are called, how long they take, and what errors occur. Implement structured logging around every tool invocation:

import logging
import time
import uuid

logger = logging.getLogger("mcp-orchestrator")

async def call_tool_with_telemetry(session, tool_name, arguments, session_id=None):
    call_id = str(uuid.uuid4())
    session_id = session_id or "unknown"
    start = time.time()
    
    logger.info(json.dumps({
        "event": "tool_call_start",
        "call_id": call_id,
        "session_id": session_id,
        "tool": tool_name,
        "arguments": arguments,
    }))
    
    try:
        result = await session.call_tool(tool_name, arguments)
        duration = time.time() - start
        logger.info(json.dumps({
            "event": "tool_call_success",
            "call_id": call_id,
            "tool": tool_name,
            "duration_ms": round(duration * 1000, 2),
        }))
        return result
    except Exception as e:
        duration = time.time() - start
        logger.error(json.dumps({
            "event": "tool_call_error",
            "call_id": call_id,
            "tool": tool_name,
            "duration_ms": round(duration * 1000, 2),
            "error": str(e),
        }))
        raise

Ship these logs to a platform like Datadog, Grafana, or CloudWatch. Key metrics to track include tool call latency, error rates, tool usage distribution, and token consumption per tool set.

Deploying MCP Servers as Microservices

For production, package each MCP server as a container and expose it over HTTP/SSE. Here is a minimal Dockerfile and deployment configuration:

# Dockerfile
FROM python:3.12-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY server.py .
EXPOSE 8080

CMD ["python", "-m", "mcp.server.sse", "--host", "0.0.0.0", "--port", "8080", "server.py:mcp"]

Requirements file:

# requirements.txt
mcp>=1.0.0
pydantic>=2.0.0

Deploy behind a reverse proxy with TLS termination, authentication, and health checks. Each server becomes independently scalable — if your search tool gets heavy traffic, scale just that server.

Best Practices Summary

• Keep tools focused — Each tool should do one thing well. Avoid "god tools" with 20 parameters.
• Write excellent docstrings — The model's ability to select the right tool depends entirely on the description.
• Use namespacing — Prefix tool names with server identity to avoid collisions in multi-server setups.
• Filter tools dynamically — Never send 100 tools to the model. Use retrieval to select a relevant subset.
• Validate all inputs — Treat model-generated arguments as untrusted user input.
• Cache idempotent calls — Significant cost and latency savings for read-heavy workloads.
• Execute independent calls concurrently — Use asyncio.gather for parallel tool execution.
• Require approval for destructive operations — Human-in-the-loop is essential for production systems.
• Log everything — Structured logging around every tool call enables debugging and auditing.
• Version your tools — When changing a tool's schema, use versioned names or maintain backward compatibility.

Conclusion

The Model Context Protocol brings much-needed standardization to function calling, making it practical to build agentic systems that span dozens of tools and services. By treating MCP servers as independently deployable microservices, dynamically filtering tools based on query relevance, enforcing security boundaries with validation and human approval, and instrumenting every call with structured telemetry, you can scale function calling from a prototype with three tools to a production system with hundreds. The key insight is that MCP is not just a protocol for connecting models to tools — it is an architecture for building composable, observable, and secure AI systems. Start simple with a single stdio server, then graduate to remote transports, connection pooling, and tool retrieval as your tool catalog and traffic grow. The investment in infrastructure pays off quickly: new capabilities become a matter of spinning up a new MCP server rather than rewriting client code.