ChaptersCircleEventsBlog
Join us for the in-person CCSK Azure course at Black Hat from August 4–5! Register now for a hands-on deep dive and secure your spot now!

A Primer on Model Context Protocol (MCP) Secure Implementation

Published 06/23/2025

A Primer on Model Context Protocol (MCP) Secure Implementation

Written by Ken Huang, CSA Fellow, Co-Chair of CSA AI Safety Working Groups and Dr. Ying-Jung Chen, Georgia Institute of Technology.

 

This implementation guide provides a comprehensive, hands-on walkthrough for building a complete system using the Model Context Protocol (MCP), a framework designed to bridge the gap between Large Language Models (LLMs) and external, real-world tools. Using a tangible use case—a 'Grid Operations Assistant'—this document details the step-by-step creation of the core MCP components. Critically, it extends beyond basic implementation to provide an essential security analysis, applying the MAESTRO agentic AI threat modeling framework to identify vulnerabilities in the MCP architecture, showcasing insecure code examples, and outlining actionable best practices. By following this guide, developers will learn not only how to construct a functional MCP application but also how to engineer it to be robust, secure, and trustworthy from the ground up.

 

1: Prerequisites

Before starting, ensure you have:

  • Python 3.11 (most stable version for AI)  installed
  • Basic understanding of asynchronous programming in Python
  • Familiarity with RESTful APIs and JSON
  • API keys for LLM providers (OpenAI, Anthropic, etc.)
  • Basic knowledge of grid operation (for the example use case)

 

2: MCP Architecture Overview

The MCP architecture consists of three main components:

  1. MCP Host: The environment where the LLM runs, responsible for interpreting the protocol and facilitating communication between the LLM and external resources.
  2. MCP Client: The application that interacts with the MCP server, sending queries and receiving responses. It manages the conversation flow and handles tool calls.
  3. MCP Server: Provides access to tools and resources that the LLM can use. It implements the protocol specification and exposes functionality through a standardized interface.

The communication flow between these components is as follows:

  1. The client initiates a connection to the MCP server.
  2. The server responds with its capabilities, including available tools and resources.
  3. The client sends a query to an LLM via the host, providing information about the available tools.
  4. The LLM processes the query and may request to use tools or access resources.
  5. The client forwards tool calls to the server.
  6. The server executes the requested tools and returns the results.
  7. The client provides the tool results back to the LLM.
  8. The LLM generates a final response based on the original query and tool results.

 

3: Setting Up the Environment

Let's start by setting up our development environment:

 

# Create a virtual environment

    python [-]m venv mcp-env
 

# Activate the virtual environment

# On Windows:

    mcp-env\Scripts\activate

# On macOS/Linux:

    source mcp-env/bin/activate

 

# Install required packages

    pip install "mcp[cli]" "aisuite[all]" python-dotenv requests pandas matplotlib numpy

 

Create a .env file to store your API keys:

 

OPENAI_API_KEY=your-openai-api-key

ANTHROPIC_API_KEY=your-anthropic-api-key

GOOGLE_API_KEY=your-google-api-key

# Add other provider keys as needed

 

4: Implementing the MCP Server

The MCP server provides tools and resources for Grid ops. Let's create a file named grid_ops_server.py.

The Grid Operations Assistant uses the FastMCP framework to provide real-time power grid management capabilities. The server exposes grid topology information and load datasets through resource endpoints, while offering three main analytical tools: analyze_load_pattern for examining electricity demand trends with rolling averages and volatility assessment, predict_outage_risk for calculating equipment failure probabilities based on weather conditions (temperature, wind, precipitation), and generate_grid_visualization for creating matplotlib charts of either hourly load profiles or regional peak load comparisons.

The system uses simulated data for different grid regions (northeast with nuclear/hydro/wind vs southwest with solar/gas/coal), processes the information using pandas for data manipulation, and can generate base64-encoded PNG visualizations for integration into monitoring dashboards, essentially providing a comprehensive MCP-based interface for grid operators to analyze power system performance and predict potential issues.

 

import json

import os

import pandas as pd

import matplotlib

import numpy as np

import io

import base64

from datetime import datetime

from typing import List, Dict, Any, Optional, Union

 

# Set matplotlib backend before importing pyplot

matplotlib.use('Agg')

import matplotlib.pyplot as plt

 

from mcp.server.fastmcp import FastMCP

 

# Initialize the MCP server

mcp = FastMCP("Grid Operations Assistant")

 

# ----- Resources -----

 

@mcp.resource("grid://topology/{region}")

def get_grid_topology(region: str) -> Dict[str, Any]:

    Retrieve power grid topology for a specific region.

    topologies = {

        "northeast": {

            "voltage_levels": [345, 138, 69],  # kV

            "substations": 45,

            "transmission_lines": 1200,  # miles

            "primary_generators": ["Nuclear", "Hydro", "Wind"]

        },

        "southwest": {

            "voltage_levels": [500, 230, 115],

            "substations": 32,

            "transmission_lines": 950,

            "primary_generators": ["Solar", "Natural Gas", "Coal"]

        }

    }

    return topologies.get(region.lower(), {"error": f"Topology for {region} not found"})

 

@mcp.resource("grid://load/{dataset_id}")

def get_grid_load_data(dataset_id: str) -> Dict[str, Any]:

    Retrieve grid load dataset by ID.

    datasets = {

        "peak_load_2023": {

            "name": "Regional Peak Load Analysis",

            "source": "NERC",

            "time_range": "2023",

            "unit": "MW",

            "data": {

                "regions": ["Northeast", "Southeast", "Midwest", "West"],

                "peak_loads": [65000, 72000, 58000, 48000]

            }

        },

        "hourly_load": {

            "name": "Hourly Load Profile",

            "source": "ISO-NE",

            "time_range": "2024-01-01 to 2024-01-07",

            "unit": "MW",

            "data": {

                "hours": list(range(24)),

                "load": [np.random.normal(15000, 2000) for _ in range(24)]

            }

        }

    }

    return datasets.get(dataset_id, {"error": f"Dataset {dataset_id} not found"})

 

# ----- Tools -----

 

@mcp.tool()

def analyze_load_pattern(dataset_id: str, window_hours: int = 24) -> Dict[str, Any]:

    Analyze load patterns in grid data.

    data = get_grid_load_data(dataset_id)

 

    if "error" in data:

        return data

 

    df = pd.DataFrame(data["data"])

    df['load'] = df['load'].rolling(window=window_hours).mean()

 

    return {

        "dataset": data["name"],

        "analysis_window": f"{window_hours}h",

        "max_load": round(df['load'].max(), 2),

        "min_load": round(df['load'].min(), 2),

        "avg_load": round(df['load'].mean(), 2),

        "trend": "stable" if df['load'].std() < 1000 else "volatile"

    }

 

@mcp.tool()

def predict_outage_risk(equipment_id: str, weather_data: Dict[str, float]) -> Dict[str, Any]:

    Predict outage risk for grid equipment based on weather conditions.

    # Simulated risk model

    base_risk = 0.05

    risk_factors = {

        "temperature": 0.001 * abs(weather_data.get("temp_c", 25) - 25),

        "wind_speed": 0.002 * weather_data.get("wind_kph", 0),

        "precipitation": 0.003 * weather_data.get("precip_mm", 0)

    }

 

    total_risk = base_risk + sum(risk_factors.values())

 

    return {

        "equipment_id": equipment_id,

        "risk_score": round(total_risk, 4),

        "risk_category": "high" if total_risk > 0.1 else "medium" if total_risk > 0.05 else "low",

        "factors": risk_factors

    }

 

@mcp.tool()

def generate_grid_visualization(dataset_id: str) -> Dict[str, Any]:

    Generate visualization of grid operational data.

    data = get_grid_load_data(dataset_id)

 

    plt.figure(figsize=(10, 6))

 

    if "hours" in data["data"]:

        plt.plot(data["data"]["hours"], data["data"]["load"], 'b-', linewidth=2)

        plt.title("Hourly Load Profile")

        plt.xlabel("Hour of Day")

        plt.ylabel("Load (MW)")

    elif "regions" in data["data"]:

        plt.bar(data["data"]["regions"], data["data"]["peak_loads"])

        plt.title("Regional Peak Loads")

        plt.ylabel("Peak Load (MW)")

 

    plt.grid(True, linestyle='--', alpha=0.7)

 

    buf = io.BytesIO()

    plt.savefig(buf, format='png', dpi=100)

    buf.seek(0)

    plt.close()

 

    return {

        "visualization": f"data:image/png;base64,{base64.b64encode(buf.read()).decode('utf-8')}",

        "dataset": data["name"]

    }

 

# ----- Server Execution -----

 

if __name__ == "__main__":

    print("GridOperationsServer:STARTED", flush=True)

    mcp.run(transport='stdio')

 

5: Implementing the MCP Host

The MCP Host is responsible for interpreting the protocol and facilitating communication between the LLM and external resources. Let's create a file named grid_ops_host.py:

This minimal MCP Host implementation serves as a lightweight bridge between users and LLM-powered tool execution. The MCPHost class initializes with an AI client and model configuration, then provides a single process_query() method that handles the complete interaction cycle: it accepts a user query and list of available MCP tools, formats the tools into the LLM's expected function calling schema, sends the query to the language model with tool definitions, and extracts any tool calls from the response along with the conversational content.

The implementation focuses purely on the essential MCP host functionality—acting as a protocol translator that converts MCP tool specifications into LLM function calls and parsing the structured responses—without maintaining conversation history or complex state management, making it ideal for understanding the fundamental architecture of how MCP servers and LLM clients communicate through standardized tool interfaces.

 

import json

import aisuite as ai

from typing import Dict, Any, List

 

class MCPHost:

    def __init__(self, model: str = "openai:gpt-4o"):

        self.client = ai.Client()

        self.model = model

 

    async def process_query(self, query: str, tools: List[Dict]) -> Dict[str, Any]:

        # Format tools for LLM

        formatted_tools = [{

            "type": "function",

            "function": {

                "name": t["name"],

                "description": t["description"],

                "parameters": t["inputSchema"]

            }

        } for t in tools]

 

        # Get LLM response

        response = self.client.chat.completions.create(

            model=self.model,

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

            tools=formatted_tools,

            temperature=0.3

        )

 

        llm_msg = response.choices[0].message

 

        # Extract tool calls if any

        tool_calls = []

        if hasattr(llm_msg, "tool_calls") and llm_msg.tool_calls:

            for tc in llm_msg.tool_calls:

                tool_calls.append({

                    "id": tc.id,

                    "name": tc.function.name,

                    "arguments": json.loads(tc.function.arguments)

                })

 

        return {

            "response": llm_msg.content,

            "tool_calls": tool_calls

        }

 

6: Implementing the MCP Client

The MCP Client connects to the server and facilitates communication between the user, the LLM, and the MCP server. Let's create a file named grid_ops_client.py:

This following MCP client implements a complete MCP  interaction system that connects to an MCP server, manages LLM conversations, and executes server-side tools seamlessly. The GridOperationsClient establishes a stdio connection to an MCP server (Python or Node.js), retrieves available tools, and runs an interactive loop where user queries are processed through the MCPHost to generate LLM responses with potential tool calls.

When the LLM requests tool execution, the client automatically calls the appropriate server tools, collects results, and feeds them back to the LLM for a final comprehensive response, creating a complete AI-powered operational assistant that can access and manipulate server-side resources through the standardized MCP protocol.

 

import sys

import json

import asyncio

import aioconsole

from contextlib import AsyncExitStack

from mcp import ClientSession, StdioServerParameters

from mcp.client.stdio import stdio_client

from grid_ops_host import MCPHost

 

class GridOperationsClient:

    """MCP Client for grid operations applications."""

 

    def __init__(self, model: str = "openai:gpt-4o"):

        self.session = None

        self.exit_stack = AsyncExitStack()

        self.host = MCPHost(model=model)

        self.tools = []

 

    async def connect_to_server(self, server_script_path: str):

        """Connect to MCP server."""

        command = "python" if server_script_path.endswith('.py') else "node"

        server_params = StdioServerParameters(command=command, args=[server_script_path])

 

        stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))

        self.stdio, self.write = stdio_transport

        self.session = await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))

 

        await self.session.initialize()

        response = await self.session.list_tools()

        self.tools = response.tools

        print(f"Connected with {len(self.tools)} tools")

 

    async def process_query(self, query: str) -> str:

        """Process operational query with tool execution."""

        # Format tools for host

        formatted_tools = [{

            "name": tool.name,

            "description": tool.description,

            "inputSchema": tool.inputSchema

        } for tool in self.tools]

 

        # Get LLM response with tool calls

        result = await self.host.process_query(query, formatted_tools)

 

        if not result["tool_calls"]:

            return result["response"]

 

        # Execute tool calls

        tool_results = []

        for tool_call in result["tool_calls"]:

            print(f"Executing: {tool_call['name']}")

 

            try:

                tool_result = await self.session.call_tool(

                    tool_call["name"], tool_call["arguments"]

                )

                tool_results.append({

                    "id": tool_call["id"],

                    "result": tool_result.content

                })

            except Exception as e:

                print(f"Tool error: {e}")

                continue

 

        # Get final answer with tool results

        return await self.host.process_tool_results(

            result["messages"], result["llm_response"], tool_results

        )

 

    async def run_loop(self):

        """Interactive operation loop."""

        print("Grid Operations Client - Type 'quit' to exit")

 

        while True:

            try:

                query = await aioconsole.ainput("\nQuery: ")

                if query.lower() == 'quit':

                    break

 

                print("Processing...")

                response = await self.process_query(query)

                print(f"\n{response}")

 

            except Exception as e:

                print(f"Error: {e}")

 

    async def cleanup(self):

        """Clean up resources."""

        await self.exit_stack.aclose()

 

async def main():

    if len(sys.argv) < 2:

        print("Usage: python client.py ")

        sys.exit(1)

 

    client = GridOperationsClient()

    try:

        await client.connect_to_server(sys.argv[1])

        await client.run_loop()

    finally:

        await client.cleanup()

 

if __name__ == "__main__":

    asyncio.run(main())

 

7: AI Suite Integration for Model Flexibility

Andrew Ng's AI Suite provides a unified interface for interacting with various LLM providers. Let's create a dedicated file to demonstrate the integration with AI Suite for model flexibility, named.

This AI Suite Manager provides a unified interface for working with multiple LLM providers (OpenAI, Anthropic, Google, Mistral) within MCP applications, handling model validation, prompt optimization, and provider-specific tool formatting automatically. The class encapsulates provider differences by optimizing system prompts with provider-specific enhancements (like adding technical analysis requests for Anthropic models), formatting tools according to each provider's expected schema (function declarations for Google, input_schema for Anthropic), and providing a single chat_completion method that abstracts away the complexity of working with different AI providers. This enables seamless model switching in MCP applications without requiring code changes, making it easy to compare performance across different LLM providers for the same operational tasks.

 

import json

from typing import Dict, Any, List, Optional

from dotenv import load_dotenv

import aisuite as ai

 

load_dotenv()

 

class AISuiteManager:

    """Manager for AI Suite integration with MCP."""

 

    def __init__(self):

        self.client = ai.Client()

        self.models = {

            "openai": ["gpt-4o", "gpt-4-turbo", "gpt-3.5-turbo"],

            "anthropic": ["claude-3-5-sonnet-20241022", "claude-3-opus-20240229"],

            "google": ["gemini-1.5-pro", "gemini-1.5-flash"],

            "mistral": ["mistral-large-latest", "mistral-medium-latest"]

        }

 

    def validate_model(self, model: str) -> bool:

        """Validate model availability."""

        if ":" not in model:

            return False

        provider, model_name = model.split(":", 1)

        return provider in self.models and model_name in self.models[provider]

 

    def optimize_prompt(self, prompt: str, model: str) -> str:

        """Optimize prompt for specific providers."""

        if ":" not in model:

            return prompt

 

        provider = model.split(":", 1)[0]

 

        optimizations = {

            "anthropic": " Include detailed technical analysis.",

            "google": " Structure response with clear sections.",

            "mistral": "",

            "openai": " Provide concise, actionable insights."

        }

 

        return prompt + optimizations.get(provider, "")

 

    def format_tools(self, tools: List[Dict[str, Any]], model: str) -> List[Dict[str, Any]]:

        """Format tools for different providers."""

        if ":" not in model or not tools:

            return tools

 

        provider = model.split(":", 1)[0]

 

        if provider == "anthropic":

            return [{

                "name": t["name"],

                "description": t["description"],

                "input_schema": t["inputSchema"]

            } for t in tools]

        elif provider == "google":

            return [{

                "function_declarations": [{

                    "name": t["name"],

                    "description": t["description"],

                    "parameters": t["inputSchema"]

                }]

            } for t in tools]

 

        return [{

            "type": "function",

            "function": {

                "name": t["name"],

                "description": t["description"],

                "parameters": t["inputSchema"]

            }

        } for t in tools]

 

    def chat_completion(self, model: str, messages: List[Dict[str, Any]],

                       tools: Optional[List[Dict[str, Any]]] = None,

                       temperature: float = 0.3) -> Any:

        """Create optimized chat completion."""

        if not self.validate_model(model):

            raise ValueError(f"Invalid model: {model}")

 

        # Optimize system prompts

        for msg in messages:

            if msg["role"] == "system":

                msg["content"] = self.optimize_prompt(msg["content"], model)

 

        return self.client.chat.completions.create(

            model=model,

            messages=messages,

            tools=self.format_tools(tools or [], model),

            temperature=temperature,

            max_tokens=2000

        )

 

Of course. I have analyzed the initial guide, the MAESTRO framework (Document 1), and the insecure code examples (Document 2). I will now extend the implementation guide by integrating a new, comprehensive section focused on security.

This new section will:

  1. Apply the MAESTRO framework to perform a threat model of the MCP architecture.
  2. Showcase the insecure code examples to illustrate concrete vulnerabilities.
  3. Provide a detailed security guide with actionable best practices and secure code patterns to mitigate the identified threats.

 

8. Securing MCP Applications: A Threat-Driven Approach

Connecting LLMs to real-world systems via MCP introduces powerful capabilities but also significant security risks. Because the LLM can autonomously decide to call tools that execute code and interact with data, a failure to secure the MCP implementation can lead to severe consequences, including data breaches, system compromise, and financial loss.

This section provides a security-focused analysis of the MCP architecture. We will use Ken Huang’s MAESTRO (Multi-Agent Environment, Security, Threat, Risk, and Outcome) framework to identify potential threats and then demonstrate how these threats manifest in poorly written code. Finally, we will provide a comprehensive guide to implementing security best practices.

 

8.1 Threat Modeling MCP with MAESTRO

The MAESTRO framework provides a layered approach to threat modeling that is well-suited for the complex, multi-component nature of an MCP system. We can map the MCP components (Server, Host, Client) to the MAESTRO layers to identify specific threats.

MAESTRO Layer

MCP Component(s)

Potential MCP-Specific Threats

Layer 7: Agent Ecosystem

Client, Host, LLM

Agent Tool Misuse & Goal Manipulation: An attacker uses prompt injection to trick the LLM into calling tools with malicious parameters. The LLM's goal is manipulated to cause harm.

Layer 6: Security & Compliance

All Components

Evasion of Security Controls: The LLM hallucinates or is tricked into generating arguments that bypass weak input validation on the MCP Server. Lack of Auditability: Poor logging makes it impossible to trace a malicious action back to a specific user query or LLM decision.

Layer 5: Evaluation & Observability

Host, Server

Poisoning Observability Data: An attacker injects false log entries by manipulating tool inputs, masking their malicious activity. Evasion of Detection: An attack is carried out through a series of seemingly benign tool calls that, when combined, are malicious.

Layer 4: Deployment & Infrastructure

Host, Server

Resource Hijacking: A vulnerable tool allows an attacker to run computationally expensive operations, leading to a Denial of Service (DoS). Information Disclosure: A tool leaks environment variables (os.environ), exposing API keys, database credentials, or other secrets.

Layer 3: Agent Frameworks

Server, Client

Supply Chain Attacks: A vulnerability in the mcp library or its dependencies (e.g., FastAPI, uvicorn) is exploited. Framework Evasion: An attacker crafts a malformed MCP message that the protocol parser handles incorrectly, leading to a crash or exploit.

Layer 2: Data Operations

MCP Server

Data Poisoning & Tampering (SQL Injection): A tool constructs SQL queries directly from user/LLM input without sanitization, allowing an attacker to modify or delete data. Data Exfiltration: A tool exposes sensitive data from a database or file system without proper authorization checks.

Layer 1: Foundation Models

LLM (via Host)

Adversarial Examples (Prompt Injection): The primary vector for initiating attacks. An attacker crafts a prompt that bypasses the LLM's safety alignment to trigger malicious tool calls. Model Stealing: An attacker could potentially use tools to infer information about the underlying model or its system prompt.

 

8.2 Insecure MCP Code Examples: What Not to Do

To demonstrate insecure MCP code, we have created a sample project on GitHub.

The following code, taken from vulnerable_mcp_sse.py, demonstrates how several of the threats identified above can be introduced through insecure coding practices.

For detailed analysis, please see DeepWiki’s link.

 

Vulnerability 1: SQL Injection in insert_record

This tool directly embeds its string arguments into an SQL query using an f-string, making it trivially vulnerable to SQL Injection.

 

@mcp.tool()

def insert_record(name: str, address: str) -> str:

    conn = sqlite3.connect(DB_NAME)

    cursor = conn.cursor()

    try:

        # VULNERABILITY: Direct string formatting into a query

        cursor.execute(f"INSERT INTO records (name, address) VALUES ('{name}', '{address}')")

        conn.commit()

        result = f"Record inserted: {name}, {address}"

    except Exception as e:

        result = f"SQL ERROR: {e}"

    conn.close()

    return result

 

  • Problem: If an attacker provides a name like test'); DROP TABLE records;--, the query becomes INSERT INTO records (name, address) VALUES ('test'); DROP TABLE records;--', 'some_address'), which will delete the records table.
  • MAESTRO Threat: Maps directly to Layer 2 (Data Operations) threats like Data Poisoning and Data Tampering.

 

Vulnerability 2: Arbitrary Code Execution in execute_sql

This tool is dangerously over-privileged, allowing any client to execute any SQL command on the database.

 

# INSECURE: Allows arbitrary SQL execution

@mcp.tool()

def execute_sql(query: str) -> str:

    conn = sqlite3.connect(DB_NAME)

    cursor = conn.cursor()

    # VULNERABILITY: Executes any query provided by the client/LLM

    result = cursor.execute(query).fetchall()

    conn.commit()

    conn.close()

    return str(result)

 

  • Problem: An attacker (or a tricked LLM) can call this tool to read sensitive data, modify schemas, or delete the entire database. There are no restrictions.
  • MAESTRO Threat: A critical example of Layer 7 (Agent Tool Misuse) and violates the principle of least privilege.

 

Vulnerability 3: Information Disclosure in get_env_variable

This tool exposes environment variables from the server, which often contain highly sensitive secrets.

 

# INSECURE: Leaks sensitive environment variables

@mcp.tool()

def get_env_variable(var_name: str) -> str:

    # VULNERABILITY: Reads and returns any environment variable

    return os.environ.get(var_name, "Not found")

 

  • Problem: An attacker can ask the LLM, "What is the OPENAI_API_KEY environment variable?" The LLM would call this tool, and the server would return the secret key.
  • MAESTRO Threat: A classic Layer 4 (Deployment & Infrastructure) threat of Information Disclosure.

 

Vulnerability 4: Lack of Authorization in query_records

This tool returns all records from the database without checking if the user making the request is authorized to see them.

 

# INSECURE: No authorization check

@mcp.tool()

def query_records() -> str:

    conn = sqlite3.connect(DB_NAME)

    cursor = conn.cursor()

    # VULNERABILITY: Returns all data to any caller

    cursor.execute("SELECT * FROM records")

    rows = cursor.fetchall()

    conn.close()

    return "\n".join([f"ID: {row[0]}, Name: {row[1]}, Address: {row[2]}" for row in rows])

 

  • Problem: In a multi-user system, any user could ask the LLM to "list all records" and gain access to data they shouldn't see.
  • MAESTRO Threat: Maps to Layer 2 (Data Exfiltration) and Layer 6 (Compliance) failures.

 

8.3 MCP Security Practices Guide

To mitigate these threats, developers must adopt a defense-in-depth strategy, securing the Server, Host, and Client components.

 

1. Secure the MCP Server: The First Line of Defense

The server is the most critical component to secure, as it directly executes code and accesses resources.

  • Use Parameterized Queries to Prevent SQL Injection: Never use string formatting (f-strings, +, %) to build SQL queries. Always use the database driver's parameter substitution mechanism.
  • Apply the Principle of Least Privilege: Tools should have the minimum permissions necessary to perform their function. Avoid creating overly powerful tools like execute_sql. Instead, create specific, constrained tools.
  • Implement Strict Input Validation: Do not trust any input from the LLM. Validate data types, lengths, ranges, and character sets. Use an allowlist for accepted values where possible.
  • Implement Authorization and Access Control: Tools that access or modify sensitive data must verify that the caller has the necessary permissions. This requires passing user identity information through the MCP call chain.
  • Do Not Expose Environment or System Internals: Never create tools like get_env_variable or execute_shell_command. Manage secrets using a secure vault (e.g., HashiCorp Vault, AWS Secrets Manager) and access them within the tool code, never exposing them to the LLM.

 

2. Harden the MCP Host: The LLM Guardian

The host sits between the LLM and the client, acting as a critical control point.

  • Sanitize and Validate LLM Output: Before sending a tool call to the client, the host should inspect the arguments generated by the LLM. Look for suspicious patterns, injection attempts, or unexpected data formats.
  • Implement a Tool Allowlist: The host should maintain an explicit list of tools the LLM is permitted to call for a given session or user. The LLM should not be able to discover or call unapproved tools.
  • Add a Human-in-the-Loop for Sensitive Operations: For high-risk tools (e.g., deleting data, transferring funds), the host should not execute the tool call immediately. Instead, it should return a confirmation request to the user via the client.

 

3. Reinforce the MCP Client: The User's Shield

The client manages the user session and orchestrates communication.

  • Maintain Session Integrity: Ensure that user sessions are managed securely and that a user cannot hijack another's session to execute tools with their privileges.
  • Enforce Final Confirmation: As mentioned above, the client is responsible for presenting confirmation dialogs to the user before executing sensitive tool calls requested by the LLM.
  • Comprehensive Logging: The client should log the entire conversation flow: user prompts, LLM responses, tool calls initiated, parameters used, and tool results received. This is crucial for auditing and incident response.

For more details on the code implementation, please visit this GitHub repository by Dr. Ying-Jung Chen.

 

9: References

1. Model Context Protocol (MCP) Official Documentation

2. MCP Python SDK

3. Andrew Ng's AI Suite

4. Grid Data Sources

5. LLM Providers

Share this content on your favorite social network today!

Unlock Cloud Security Insights

Unlock Cloud Security Insights

Choose the CSA newsletters that match your interests:

Subscribe to our newsletter for the latest expert trends and updates