Introduction

Large Language Models generate text from the information available in their context. They do not automatically have access to current databases, organizational documents, calculators, business applications, or external APIs.

Application developers can connect models to these systems through custom integrations. However, building a different connector for every combination of AI application and external system creates duplicated implementation, inconsistent interfaces, and additional security and maintenance responsibilities.

The Model Context Protocol, abbreviated as MCP, provides a standardized protocol through which compatible AI applications can connect to servers that expose context and capabilities.

Tool calling allows a model to request a structured operation such as a calculation, search, database lookup, or API call. The application remains responsible for validating, authorizing, executing, and auditing the requested operation.

Learning Objectives

Prerequisites

python -m pip install \
  mcp \
  pydantic \
  httpx \
  python-dotenv \
  pytest

What Is MCP?

MCP is an open protocol that defines a standard way for an AI application to obtain contextual information and interact with external capabilities exposed by compatible servers.

MCP Architecture

MCP Host

The host is the AI application that coordinates user interaction, model access, MCP clients, permissions, consent, and result presentation.

MCP Client

An MCP client is a protocol connector managed by the host. It establishes communication with an MCP server and exchanges protocol messages.

MCP Server

An MCP server exposes a focused set of approved capabilities. It translates protocol operations into interactions with local or remote systems.

Core MCP Primitives

MCP Tools

An MCP tool is a named capability with a description and structured input schema. The description helps the model or application understand when the tool is appropriate.

{
    "name": "calculate_order_total",
    "description": "Calculate the total of non-negative item prices.",
    "inputSchema": {
        "type": "object",
        "properties": {
            "item_prices": {
                "type": "array",
                "items": {
                    "type": "number",
                    "minimum": 0
                },
                "minItems": 1
            }
        },
        "required": [
            "item_prices"
        ],
        "additionalProperties": false
    }
}

MCP Resources

Resources expose contextual information that an application can discover or read. A resource is commonly identified through a URI.

{
    "uri": "knowledge://policies/password-reset",
    "name": "Password Reset Policy",
    "description": "Approved password-reset policy content.",
    "mimeType": "text/plain"
}

MCP Prompts

MCP prompts are reusable prompt templates exposed by a server. They can standardize interactions with a capability or domain.

{
    "name": "summarize_policy",
    "description": "Create a concise summary of an approved policy resource.",
    "arguments": [
        {
            "name": "policy_uri",
            "description": "URI of the authorized policy resource.",
            "required": true
        },
        {
            "name": "audience",
            "description": "Intended audience for the summary.",
            "required": true
        }
    ]
}

JSON-RPC Communication

MCP uses JSON-RPC messages. A request includes a method and may include parameters. A response contains either a result or an error associated with the request identifier.

{
    "jsonrpc": "2.0",
    "id": 1,
    "method": "tools/list",
    "params": {}
}

{
    "jsonrpc": "2.0",
    "id": 1,
    "result": {
        "tools": []
    }
}

Initialization and Capability Negotiation

Before ordinary operations begin, the client and server establish protocol compatibility and exchange information about supported capabilities.

Tool Discovery

The host can make authorized tool definitions available to the model. The model may then request one of those tools using arguments that should conform to the tool's schema.

Tool Invocation Lifecycle

Build a Basic MCP Server

from mcp.server.fastmcp import FastMCP


mcp_server = FastMCP(
    "Course Utility Server"
)


@mcp_server.tool()
def calculate_order_total(
    item_prices: list[
        float
    ]
) -> dict:
    """Calculate the total of non-negative item prices."""

    if not item_prices:

        raise ValueError(
            "At least one item price is required."
        )

    if any(
        item_price < 0
        for item_price in item_prices
    ):
        raise ValueError(
            "Item prices cannot be negative."
        )

    total = sum(
        item_prices
    )

    return {
        "item_count": len(
            item_prices
        ),
        "total": round(
            total,
            2
        )
    }


if __name__ == "__main__":

    mcp_server.run()

Expose a Resource

from mcp.server.fastmcp import FastMCP


mcp_server = FastMCP(
    "Knowledge Server"
)


APPROVED_POLICY_TEXT = (
    "Password resets must use the "
    "approved identity-verification workflow."
)


@mcp_server.resource(
    "knowledge://policies/password-reset"
)
def read_password_reset_policy() -> str:
    """Return the approved password-reset policy."""

    return APPROVED_POLICY_TEXT

Expose a Prompt

@mcp_server.prompt()
def summarize_policy(
    policy_text: str,
    audience: str
) -> str:
    """Create a controlled policy-summary prompt."""

    return (
        "Summarize the supplied policy for "
        f"the following audience: {audience}.\n\n"
        "Use only the supplied policy. "
        "Do not add requirements.\n\n"
        f"Policy:\n{policy_text}"
    )

Build a Tool that Calls an External API

import os

import httpx
from mcp.server.fastmcp import FastMCP


mcp_server = FastMCP(
    "Approved Service Connector"
)


@mcp_server.tool()
async def get_record_status(
    record_id: str
) -> dict:
    """Get the status of an authorized record."""

    normalized_record_id = (
        record_id.strip()
    )

    if not normalized_record_id:

        raise ValueError(
            "record_id is required."
        )

    if len(
        normalized_record_id
    ) > 100:

        raise ValueError(
            "record_id exceeds the "
            "approved length."
        )

    service_base_url = os.environ[
        "APPROVED_SERVICE_BASE_URL"
    ]

    access_token = os.environ[
        "APPROVED_SERVICE_ACCESS_TOKEN"
    ]

    request_url = (
        f"{service_base_url}/records/"
        f"{normalized_record_id}"
    )

    timeout = httpx.Timeout(
        10.0
    )

    async with httpx.AsyncClient(
        timeout=timeout,
        verify=True
    ) as client:

        response = await client.get(
            request_url,
            headers={
                "Authorization": (
                    f"Bearer {access_token}"
                )
            }
        )

        response.raise_for_status()

        response_data = response.json()

    return {
        "record_id": normalized_record_id,
        "status": response_data.get(
            "status"
        )
    }

Create a Read-Only Database Tool

import os

import sqlite3
from mcp.server.fastmcp import FastMCP


mcp_server = FastMCP(
    "Inventory Lookup Server"
)


@mcp_server.tool()
def get_inventory_summary(
    item_id: str
) -> dict:
    """Get an inventory summary for one item."""

    normalized_item_id = (
        item_id.strip()
    )

    if not normalized_item_id:

        raise ValueError(
            "item_id is required."
        )

    database_path = os.environ[
        "APPROVED_INVENTORY_DATABASE"
    ]

    query = (
        "SELECT item_id, available_quantity "
        "FROM inventory_summary "
        "WHERE item_id = ?"
    )

    with sqlite3.connect(
        database_path
    ) as connection:

        connection.row_factory = (
            sqlite3.Row
        )

        result = connection.execute(
            query,
            (
                normalized_item_id,
            )
        ).fetchone()

    if result is None:

        return {
            "found": False,
            "item_id": normalized_item_id
        }

    return {
        "found": True,
        "item_id": result[
            "item_id"
        ],
        "available_quantity": result[
            "available_quantity"
        ]
    }

Direct Tool Calling vs MCP

Generic Model Tool-Calling Loop

def run_tool_calling_loop(
    model_client,
    messages,
    available_tools,
    execute_tool,
    maximum_tool_calls=5
):
    tool_call_count = 0

    while True:

        model_response = model_client.generate(
            messages=messages,
            tools=available_tools
        )

        tool_requests = (
            model_response.get(
                "tool_calls",
                []
            )
        )

        if not tool_requests:

            return model_response[
                "content"
            ]

        for tool_request in tool_requests:

            tool_call_count += 1

            if (
                tool_call_count
                >
                maximum_tool_calls
            ):
                raise RuntimeError(
                    "Maximum tool-call limit reached."
                )

            tool_name = tool_request[
                "name"
            ]

            tool_arguments = tool_request[
                "arguments"
            ]

            tool_result = execute_tool(
                tool_name=tool_name,
                arguments=tool_arguments
            )

            messages.append(
                {
                    "role": "tool",
                    "name": tool_name,
                    "content": tool_result
                }
            )

Validate Tool Arguments

from pydantic import BaseModel
from pydantic import Field
from pydantic import ValidationError


class OrderTotalInput(
    BaseModel
):
    item_prices: list[
        float
    ] = Field(
        min_length=1,
        max_length=100
    )


def validate_order_total_arguments(
    proposed_arguments
):
    try:
        validated_arguments = (
            OrderTotalInput.model_validate(
                proposed_arguments
            )
        )

        if any(
            item_price < 0
            for item_price
            in validated_arguments.item_prices
        ):
            raise ValueError(
                "Negative item prices "
                "are not permitted."
            )

        return {
            "valid": True,
            "arguments": (
                validated_arguments.model_dump()
            ),
            "error": None
        }

    except (
        ValidationError,
        ValueError
    ) as error:

        return {
            "valid": False,
            "arguments": None,
            "error": str(
                error
            )
        }

Authentication vs Authorization

Human Approval

Consequential tools should require clear user confirmation before execution.

Read Tools vs Action Tools

Idempotency

An idempotent operation can be repeated with the same idempotency identity without producing an additional unintended effect.

from pydantic import BaseModel
from pydantic import Field


class CreateRequest(
    BaseModel
):
    idempotency_key: str = Field(
        min_length=16,
        max_length=128
    )

    requested_value: str = Field(
        min_length=1,
        max_length=500
    )


def create_record_safely(
    request: CreateRequest,
    repository
):
    previous_result = (
        repository.find_by_idempotency_key(
            request.idempotency_key
        )
    )

    if previous_result is not None:

        return previous_result

    return repository.create(
        idempotency_key=(
            request.idempotency_key
        ),
        requested_value=(
            request.requested_value
        )
    )

Tool Error Handling

Timeouts and Bounded Retries

import asyncio


async def run_with_timeout(
    operation,
    timeout_seconds
):
    if timeout_seconds <= 0:

        raise ValueError(
            "timeout_seconds must be positive."
        )

    try:
        return await asyncio.wait_for(
            operation,
            timeout=timeout_seconds
        )

    except asyncio.TimeoutError as error:

        raise TimeoutError(
            "Tool execution exceeded "
            "the approved timeout."
        ) from error

Prompt Injection and Tool Poisoning

Untrusted content may attempt to influence a model into calling tools, disclosing data, or ignoring application policy. Tool names, descriptions, results, resources, and prompts can also become attack surfaces if their source or integrity is not controlled.

Supply-Chain Security

Test MCP Tools Directly

def test_calculate_order_total():

    result = calculate_order_total(
        item_prices=[
            12.50,
            7.25,
            3.00
        ]
    )

    assert result[
        "item_count"
    ] == 3

    assert result[
        "total"
    ] == 22.75


def test_calculate_order_total_rejects_negative():

    try:
        calculate_order_total(
            item_prices=[
                10.00,
                -2.00
            ]
        )

        assert False

    except ValueError as error:

        assert (
            "negative"
            in str(
                error
            ).lower()
        )

MCP Testing Layers

MCP Observability

MCP Metrics

Version the Complete MCP Integration

Responsible Tool Calling

Common MCP and Tool-Calling Mistakes

Troubleshooting

Recommended MCP Workflow

Hands-On Practice

Knowledge Check

MCP Readiness Checklist

Lesson Summary