MCP Protocol and AI Agent Toolchains: A Developer's Essential Guide for 2026

+-------------------+     MCP Protocol     +-------------------+
|   MCP Client      | <-----------------> |   MCP Server      |
|  (Claude/GPT)     |                     |  (Tool Provider)   |
+-------------------+                     +-------------------+
       |                                         |
       v                                         v
  Model Inference                         File System/API/DB

# Install MCP SDK (Python)
pip install mcp anthropic

# Or use TypeScript/Node.js
npm install @modelcontextprotocol/sdk @anthropic-ai/sdk

import os
import json
from mcp.server import Server
from mcp.types import Tool, TextContent
import mcp.server.stdio

# Create MCP Server instance
server = Server("filesystem-server")

# Define tool list
@server.list_tools()
async def list_tools():
    return [
        Tool(
            name="read_file",
            description="Read the contents of a file at the specified path",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {
                        "type": "string",
                        "description": "Absolute path to the file"
                    }
                },
                "required": ["path"]
            }
        ),
        Tool(
            name="write_file",
            description="Write content to a file at the specified path",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {
                        "type": "string",
                        "description": "Absolute path to the file"
                    },
                    "content": {
                        "type": "string",
                        "description": "Content to write"
                    }
                },
                "required": ["path", "content"]
            }
        ),
        Tool(
            name="list_directory",
            description="List all files and subdirectories in a directory",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {
                        "type": "string",
                        "description": "Directory path"
                    }
                },
                "required": ["path"]
            }
        )
    ]

# Implement tool call logic
@server.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "read_file":
        path = arguments["path"]
        if not os.path.exists(path):
            return [TextContent(type="text", text=f"Error: File {path} does not exist")]
        with open(path, "r", encoding="utf-8") as f:
            content = f.read()
        return [TextContent(type="text", text=content)]

    elif name == "write_file":
        path = arguments["path"]
        content = arguments["content"]
        os.makedirs(os.path.dirname(path), exist_ok=True)
        with open(path, "w", encoding="utf-8") as f:
            f.write(content)
        return [TextContent(type="text", text=f"Successfully wrote file: {path}")]

    elif name == "list_directory":
        path = arguments["path"]
        entries = os.listdir(path)
        result = []
        for entry in entries:
            full_path = os.path.join(path, entry)
            entry_type = "[DIR]" if os.path.isdir(full_path) else "[FILE]"
            result.append(f"{entry_type} {entry}")
        return [TextContent(type="text", text="\n".join(result))]

# Start the server
async def main():
    async with mcp.server.stdio.stdio_server() as (read, write):
        await server.run(read, write, server.create_initialization_options())

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import Database from "better-sqlite3";

const server = new Server(
  { name: "sqlite-server", version: "1.0.0" },
  { capabilities: { tools: {} } }
);

// Register tool list
server.setRequestHandler("tools/list", async () => ({
  tools: [
    {
      name: "query_database",
      description: "Execute SQL query and return results (read-only)",
      inputSchema: {
        type: "object",
        properties: {
          sql: { type: "string", description: "SQL query statement" },
          database: { type: "string", description: "Database file path" }
        },
        required: ["sql", "database"]
      }
    },
    {
      name: "list_tables",
      description: "List all tables in the database",
      inputSchema: {
        type: "object",
        properties: {
          database: { type: "string", description: "Database file path" }
        },
        required: ["database"]
      }
    }
  ]
}));

// Handle tool calls
server.setRequestHandler("tools/call", async (request) => {
  const { name, arguments: args } = request.params;

  if (name === "query_database") {
    const db = new Database(args.database as string, { readonly: true });
    try {
      // Security check: only allow SELECT statements
      const sql = (args.sql as string).trim().toUpperCase();
      if (!sql.startsWith("SELECT")) {
        return {
          content: [{ type: "text", text: "Error: Only SELECT queries are allowed" }]
        };
      }
      const rows = db.prepare(args.sql as string).all();
      return {
        content: [{ type: "text", text: JSON.stringify(rows, null, 2) }]
      };
    } finally {
      db.close();
    }
  }

  if (name === "list_tables") {
    const db = new Database(args.database as string, { readonly: true });
    try {
      const tables = db.prepare(
        "SELECT name FROM sqlite_master WHERE type='table'"
      ).all();
      return {
        content: [{ type: "text", text: JSON.stringify(tables, null, 2) }]
      };
    } finally {
      db.close();
    }
  }
});

// Start the server
const transport = new StdioServerTransport();
await server.connect(transport);

import anthropic
import json

client = anthropic.Anthropic(api_key="your-api-key")

# Define multiple tools
tools = [
    {
        "name": "web_search",
        "description": "Search the internet for latest information",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "Search keywords"}
            },
            "required": ["query"]
        }
    },
    {
        "name": "execute_code",
        "description": "Execute Python code and return results",
        "input_schema": {
            "type": "object",
            "properties": {
                "code": {"type": "string", "description": "Python code to execute"}
            },
            "required": ["code"]
        }
    },
    {
        "name": "create_chart",
        "description": "Generate visualization charts from data",
        "input_schema": {
            "type": "object",
            "properties": {
                "data": {"type": "object", "description": "Chart data"},
                "chart_type": {
                    "type": "string",
                    "enum": ["bar", "line", "pie", "scatter"],
                    "description": "Type of chart"
                },
                "title": {"type": "string", "description": "Chart title"}
            },
            "required": ["data", "chart_type", "title"]
        }
    }
]

# Implement tool call handlers
def handle_tool_call(name, input_data):
    if name == "web_search":
        return f"Search results for '{input_data['query']}'..."
    elif name == "execute_code":
        try:
            exec_globals = {}
            exec(input_data["code"], exec_globals)
            return "Code executed successfully"
        except Exception as e:
            return f"Execution error: {str(e)}"
    elif name == "create_chart":
        return f"Generated {input_data['chart_type']} chart: {input_data['title']}"

# Agent loop: automatically handle multi-turn tool calls
def run_agent(user_message, max_iterations=10):
    messages = [{"role": "user", "content": user_message}]

    for i in range(max_iterations):
        response = client.messages.create(
            model="claude-4-7-sonnet-20260514",
            max_tokens=4096,
            tools=tools,
            messages=messages
        )

        # Check if tool calls are needed
        if response.stop_reason == "tool_use":
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    result = handle_tool_call(block.name, block.input)
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": result
                    })

            messages.append({"role": "assistant", "content": response.content})
            messages.append({"role": "user", "content": tool_results})
        else:
            return response.content[0].text

    return "Agent stopped: max iterations reached"

# Usage example
result = run_agent("Analyze Q1 2026 AI industry data and generate comparison charts")
print(result)

from openai import OpenAI

client = OpenAI(api_key="your-api-key")

# GPT-5.5 supports strict mode tool definitions
tools = [
    {
        "type": "function",
        "function": {
            "name": "analyze_code",
            "description": "Analyze code quality and potential issues",
            "strict": True,  # New in GPT-5.5: strict mode ensures output format
            "parameters": {
                "type": "object",
                "properties": {
                    "language": {
                        "type": "string",
                        "enum": ["python", "javascript", "typescript", "go", "rust"]
                    },
                    "code": {"type": "string"},
                    "checks": {
                        "type": "array",
                        "items": {
                            "type": "string",
                            "enum": ["security", "performance", "style", "bugs"]
                        }
                    }
                },
                "required": ["language", "code", "checks"],
                "additionalProperties": False
            }
        }
    }
]

# GPT-5.5 parallel tool calling
response = client.chat.completions.create(
    model="gpt-5.5-turbo",
    messages=[
        {"role": "system", "content": "You are a code review expert."},
        {"role": "user", "content": "Review this code for security and performance."}
    ],
    tools=tools,
    tool_choice="auto",
    parallel_tool_calls=True  # GPT-5.5 supports parallel tool calls
)

# Handle parallel tool call results
for tool_call in response.choices[0].message.tool_calls:
    func_name = tool_call.function.name
    func_args = json.loads(tool_call.function.arguments)
    print(f"Tool called: {func_name}")
    print(f"Arguments: {json.dumps(func_args, indent=2)}")

import asyncio
from dataclasses import dataclass
from typing import Callable, Any

@dataclass
class AgentStep:
    """Agent execution step"""
    name: str
    tool: str
    description: str
    handler: Callable

class AIAgentWorkflow:
    """AI Agent workflow engine"""

    def __init__(self, model: str = "claude-4-7-sonnet-20260514"):
        self.model = model
        self.steps: list[AgentStep] = []
        self.context: dict[str, Any] = {}
        self.client = anthropic.Anthropic()

    def add_step(self, step: AgentStep):
        self.steps.append(step)
        return self

    async def execute(self, initial_input: str) -> dict:
        """Execute the complete workflow"""
        self.context["input"] = initial_input
        results = {}

        for step in self.steps:
            print(f"Executing step: {step.name} - {step.description}")

            prompt = f"""
            Current step: {step.description}
            Available tool: {step.tool}
            Context: {json.dumps(self.context, ensure_ascii=False)}
            Input: {initial_input}
            Please decide how to execute this step and output tool call parameters.
            """

            response = self.client.messages.create(
                model=self.model,
                max_tokens=2048,
                messages=[{"role": "user", "content": prompt}]
            )

            step_result = await step.handler(response.content[0].text)
            results[step.name] = step_result
            self.context[step.name] = step_result

            print(f"Step completed: {step.name}")

        return results

# Usage example: Code review workflow
async def code_review_workflow():
    workflow = AIAgentWorkflow()

    workflow.add_step(AgentStep(
        name="fetch_code",
        tool="git_clone",
        description="Fetch code from Git repository",
        handler=lambda x: {"files": ["main.py", "utils.py", "config.py"]}
    ))

    workflow.add_step(AgentStep(
        name="static_analysis",
        tool="pylint",
        description="Run static code analysis",
        handler=lambda x: {"issues": 3, "score": 8.5}
    ))

    workflow.add_step(AgentStep(
        name="ai_review",
        tool="claude_review",
        description="Deep code review using Claude",
        handler=lambda x: {"suggestions": ["Add type annotations", "Improve error handling"]}
    ))

    workflow.add_step(AgentStep(
        name="generate_report",
        tool="report_generator",
        description="Generate review report",
        handler=lambda x: {"report_url": "/reports/review-2026-05-08.html"}
    ))

    results = await workflow.execute("Review project: my-web-app")
    print(f"Review results: {json.dumps(results, indent=2)}")

# Run the workflow
asyncio.run(code_review_workflow())

{
  "mcpServers": {
    "filesystem": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-filesystem", "/home/user/projects"]
    },
    "postgres": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-postgres", "postgresql://localhost/mydb"]
    },
    "github": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-github"],
      "env": {
        "GITHUB_TOKEN": "ghp_xxxxxxxxxxxx"
      }
    }
  }
}

# Wrong: granting excessive permissions
@server.call_tool()
async def call_tool(name, arguments):
    if name == "execute_command":
        os.system(arguments["command"])  # Dangerous!

# Right: limit executable commands
ALLOWED_COMMANDS = {"ls", "cat", "wc", "head", "tail"}

@server.call_tool()
async def call_tool(name, arguments):
    if name == "execute_command":
        cmd = arguments["command"].split()[0]
        if cmd not in ALLOWED_COMMANDS:
            return [TextContent(type="text", text=f"Command {cmd} is not in allowlist")]
        result = subprocess.run(arguments["command"], capture_output=True, text=True)
        return [TextContent(type="text", text=result.stdout)]

import shlex
import subprocess

def safe_execute(command: str, timeout: int = 30) -> str:
    """Safely execute commands with timeout and output limits"""
    try:
        args = shlex.split(command)
        result = subprocess.run(
            args,
            capture_output=True,
            text=True,
            timeout=timeout,
            cwd="/tmp/sandbox"  # Restrict working directory
        )
        output = result.stdout[:10000]
        if result.stderr:
            output += f"\nSTDERR: {result.stderr[:5000]}"
        return output
    except subprocess.TimeoutExpired:
        return "Error: command execution timed out"
    except Exception as e:
        return f"Execution error: {str(e)}"

import logging
from datetime import datetime

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

@server.call_tool()
async def call_tool(name: str, arguments: dict):
    start_time = datetime.now()
    logger.info(f"Tool call started: {name}, args: {json.dumps(arguments)}")

    try:
        result = await _handle_tool(name, arguments)
        duration = (datetime.now() - start_time).total_seconds()
        logger.info(f"Tool call completed: {name}, duration: {duration:.2f}s")
        return result
    except Exception as e:
        logger.error(f"Tool call failed: {name}, error: {str(e)}")
        raise