MCP Protocol and Advanced Tool Orchestration

Overview

Imagine trying to organize a global conference where speakers, venues, catering, and technology all need to work together seamlessly. Without standardized protocols—common ways to communicate, share information, and coordinate—the event would be chaos. Everyone would be speaking different languages, using incompatible formats, and working at cross purposes.

This is exactly the challenge that AI agents face when trying to use tools from different providers, frameworks, and systems. The Model Context Protocol (MCP) solves this by providing a standardized way for AI systems to securely and efficiently access external tools and data sources.

In this lesson, we'll explore how MCP transforms tool integration from ad-hoc connections into robust, scalable ecosystems where agents can discover, trust, and orchestrate sophisticated tool workflows.

Learning Objectives

After completing this lesson, you will be able to:

• Understand the Model Context Protocol (MCP) architecture and benefits
• Implement MCP-compliant tools and clients
• Design secure, scalable tool ecosystems using MCP standards
• Build advanced tool orchestration patterns with dependency management
• Handle complex workflows with multiple tools and data flows

The Challenge of Tool Fragmentation

Before MCP: The Wild West of Tool Integration

Before standardized protocols, each AI framework had its own way of integrating tools:

LangChain Tools:

Copy
from langchain.tools import BaseTool

class CustomTool(BaseTool):
    name = "my_tool"
    description = "A custom tool"
    
    def _run(self, query: str) -> str:
        return "Tool result"

OpenAI Function Calling:

Copy
{
  "name": "my_tool",
  "description": "A custom tool",
  "parameters": {
    "type": "object",
    "properties": {
      "query": {"type": "string"}
    }
  }
}

Anthropic Tool Use:

Copy
<tool_description>
<tool_name>my_tool</tool_name>
<description>A custom tool</description>
<parameters>
<parameter name="query" type="string">Query to process</parameter>
</parameters>
</tool_description>

Each approach required different implementations, making it difficult to:

• Share tools across platforms
• Ensure security and reliability
• Manage complex tool dependencies
• Scale tool ecosystems

MCP: A Universal Standard

The Model Context Protocol provides:

• Standardized Interface: Common way to describe and invoke tools
• Security Model: Built-in authentication and authorization
• Discovery Mechanism: Agents can find available tools dynamically
• Dependency Management: Handle complex tool relationships
• Error Handling: Consistent error reporting and recovery

Understanding MCP Architecture

Core Components

MCP defines several key components that work together:

1. MCP Server (Tool Provider):

• Hosts and exposes tools
• Handles authentication and authorization
• Manages tool lifecycle and dependencies

2. MCP Client (AI Agent):

• Discovers and connects to MCP servers
• Invokes tools through standardized interface
• Manages tool results and error handling

3. MCP Protocol:

• Defines communication format (JSON-RPC)
• Specifies authentication mechanisms
• Handles capability negotiation

Copy
# MCP Protocol Implementation Example
import json
import asyncio
from typing import Dict, Any, List, Optional
from dataclasses import dataclass
from enum import Enum

class MCPMessageType(Enum):
    """MCP message types"""
    REQUEST = "request"
    RESPONSE = "response"
    NOTIFICATION = "notification"

@dataclass
class MCPMessage:
    """Base MCP message structure"""
    jsonrpc: str = "2.0"
    id: Optional[str] = None
    method: Optional[str] = None
    params: Optional[Dict] = None
    result: Optional[Any] = None
    error: Optional[Dict] = None

class MCPServer:
    """MCP Server implementation"""
    
    def __init__(self, name: str, version: str):
        self.name = name
        self.version = version
        self.tools = {}
        self.resources = {}
        self.capabilities = {
            "tools": True,
            "resources": True,
            "logging": True,
            "prompts": False
        }
    
    def register_tool(self, tool_spec: Dict):
        """Register a tool with the MCP server"""
        tool_name = tool_spec["name"]
        self.tools[tool_name] = tool_spec
        print(f"Registered tool: {tool_name}")
    
    def register_resource(self, resource_spec: Dict):
        """Register a resource with the MCP server"""
        resource_uri = resource_spec["uri"]
        self.resources[resource_uri] = resource_spec
        print(f"Registered resource: {resource_uri}")
    
    async def handle_initialize(self, params: Dict) -> Dict:
        """Handle MCP initialize request"""
        client_info = params.get("clientInfo", {})
        print(f"Client connecting: {client_info.get('name', 'Unknown')}")
        
        return {
            "serverInfo": {
                "name": self.name,
                "version": self.version
            },
            "capabilities": self.capabilities
        }
    
    async def handle_tools_list(self, params: Dict) -> Dict:
        """Handle tools/list request"""
        return {
            "tools": [
                {
                    "name": name,
                    "description": spec["description"],
                    "inputSchema": spec["inputSchema"]
                }
                for name, spec in self.tools.items()
            ]
        }
    
    async def handle_tools_call(self, params: Dict) -> Dict:
        """Handle tools/call request"""
        tool_name = params["name"]
        arguments = params.get("arguments", {})
        
        if tool_name not in self.tools:
            raise Exception(f"Tool '{tool_name}' not found")
        
        # Execute the tool
        tool_spec = self.tools[tool_name]
        result = await self.execute_tool(tool_spec, arguments)
        
        return {
            "content": [
                {
                    "type": "text",
                    "text": str(result)
                }
            ]
        }
    
    async def execute_tool(self, tool_spec: Dict, arguments: Dict) -> Any:
        """Execute a tool with given arguments"""
        # This would call the actual tool implementation
        tool_function = tool_spec["function"]
        return await tool_function(**arguments)
    
    async def handle_request(self, message: MCPMessage) -> MCPMessage:
        """Handle incoming MCP request"""
        try:
            if message.method == "initialize":
                result = await self.handle_initialize(message.params or {})
            elif message.method == "tools/list":
                result = await self.handle_tools_list(message.params or {})
            elif message.method == "tools/call":
                result = await self.handle_tools_call(message.params or {})
            else:
                raise Exception(f"Unknown method: {message.method}")
            
            return MCPMessage(
                id=message.id,
                result=result
            )
            
        except Exception as e:
            return MCPMessage(
                id=message.id,
                error={
                    "code": -32603,
                    "message": str(e)
                }
            )

class MCPClient:
    """MCP Client implementation"""
    
    def __init__(self, name: str, version: str):
        self.name = name
        self.version = version
        self.servers = {}
        self.capabilities = {
            "tools": True,
            "resources": True
        }
    
    async def connect_to_server(self, server_url: str) -> str:
        """Connect to an MCP server"""
        # Initialize connection
        init_message = MCPMessage(
            id="init-1",
            method="initialize",
            params={
                "clientInfo": {
                    "name": self.name,
                    "version": self.version
                },
                "capabilities": self.capabilities
            }
        )
        
        # Send initialize request (simplified - would use actual transport)
        response = await self.send_message(server_url, init_message)
        
        if response.error:
            raise Exception(f"Failed to connect: {response.error}")
        
        server_id = f"server_{len(self.servers)}"
        self.servers[server_id] = {
            "url": server_url,
            "info": response.result["serverInfo"],
            "capabilities": response.result["capabilities"]
        }
        
        return server_id
    
    async def list_tools(self, server_id: str) -> List[Dict]:
        """List available tools from a server"""
        server = self.servers[server_id]
        
        message = MCPMessage(
            id=f"tools-list-{server_id}",
            method="tools/list"
        )
        
        response = await self.send_message(server["url"], message)
        
        if response.error:
            raise Exception(f"Failed to list tools: {response.error}")
        
        return response.result["tools"]
    
    async def call_tool(self, server_id: str, tool_name: str, arguments: Dict) -> Any:
        """Call a tool on a specific server"""
        server = self.servers[server_id]
        
        message = MCPMessage(
            id=f"tool-call-{tool_name}",
            method="tools/call",
            params={
                "name": tool_name,
                "arguments": arguments
            }
        )
        
        response = await self.send_message(server["url"], message)
        
        if response.error:
            raise Exception(f"Tool call failed: {response.error}")
        
        return response.result["content"]
    
    async def send_message(self, server_url: str, message: MCPMessage) -> MCPMessage:
        """Send a message to an MCP server (simplified)"""
        # In a real implementation, this would use WebSocket, HTTP, or other transport
        # For demo purposes, we'll simulate the response
        
        if message.method == "initialize":
            return MCPMessage(
                id=message.id,
                result={
                    "serverInfo": {"name": "demo-server", "version": "1.0.0"},
                    "capabilities": {"tools": True, "resources": True}
                }
            )
        elif message.method == "tools/list":
            return MCPMessage(
                id=message.id,
                result={
                    "tools": [
                        {
                            "name": "calculate",
                            "description": "Perform calculations",
                            "inputSchema": {
                                "type": "object",
                                "properties": {
                                    "expression": {"type": "string"}
                                }
                            }
                        }
                    ]
                }
            )
        elif message.method == "tools/call":
            return MCPMessage(
                id=message.id,
                result={
                    "content": [
                        {"type": "text", "text": "Calculation result: 42"}
                    ]
                }
            )

# Example usage
async def mcp_demo():
    """Demonstrate MCP client-server interaction"""
    
    # Create MCP client
    client = MCPClient("demo-client", "1.0.0")
    
    # Connect to server
    server_id = await client.connect_to_server("http://localhost:8080")
    print(f"Connected to server: {server_id}")
    
    # List available tools
    tools = await client.list_tools(server_id)
    print(f"Available tools: {[tool['name'] for tool in tools]}")
    
    # Call a tool
    result = await client.call_tool(server_id, "calculate", {"expression": "2 + 2"})
    print(f"Tool result: {result}")

# Run the demo
# asyncio.run(mcp_demo())

MCP Message Flow

The typical MCP interaction follows this pattern:

1. Connection & Initialization

   Copy
   Client -> Server: {"method": "initialize", "params": {...}}
   Server -> Client: {"result": {"serverInfo": {...}, "capabilities": {...}}}

2. Tool Discovery

   Copy
   Client -> Server: {"method": "tools/list"}
   Server -> Client: {"result": {"tools": [...]}}

3. Tool Execution

   Copy
   Client -> Server: {"method": "tools/call", "params": {"name": "...", "arguments": {...}}}
   Server -> Client: {"result": {"content": [...]}}

Advanced Tool Orchestration Patterns

1. Tool Chaining and Workflows

MCP enables sophisticated tool orchestration where the output of one tool becomes the input to another:

Copy
# Advanced Tool Orchestration with MCP
from typing import List, Dict, Any
from dataclasses import dataclass
import asyncio

@dataclass
class WorkflowStep:
    """Represents a step in a tool workflow"""
    tool_name: str
    server_id: str
    arguments: Dict
    depends_on: List[str] = None  # List of previous step IDs
    step_id: str = None

class MCPOrchestrator:
    """Orchestrates complex workflows across multiple MCP servers"""
    
    def __init__(self, client: MCPClient):
        self.client = client
        self.workflows = {}
        self.step_results = {}
    
    async def execute_workflow(self, workflow_id: str, steps: List[WorkflowStep]) -> Dict:
        """Execute a multi-step workflow"""
        self.workflows[workflow_id] = steps
        self.step_results[workflow_id] = {}
        
        # Build dependency graph
        dependency_graph = self.build_dependency_graph(steps)
        
        # Execute steps in topological order
        execution_order = self.topological_sort(dependency_graph)
        
        for step_id in execution_order:
            step = next(s for s in steps if s.step_id == step_id)
            result = await self.execute_step(workflow_id, step)
            self.step_results[workflow_id][step_id] = result
        
        return self.step_results[workflow_id]
    
    def build_dependency_graph(self, steps: List[WorkflowStep]) -> Dict:
        """Build a dependency graph for workflow steps"""
        graph = {}
        
        for step in steps:
            if not step.step_id:
                step.step_id = f"step_{steps.index(step)}"
            
            graph[step.step_id] = step.depends_on or []
        
        return graph
    
    def topological_sort(self, graph: Dict) -> List[str]:
        """Perform topological sort to determine execution order"""
        in_degree = {node: 0 for node in graph}
        
        # Calculate in-degrees
        for node in graph:
            for neighbor in graph[node]:
                if neighbor in in_degree:
                    in_degree[neighbor] += 1
        
        # Find nodes with no incoming edges
        queue = [node for node in in_degree if in_degree[node] == 0]
        result = []
        
        while queue:
            node = queue.pop(0)
            result.append(node)
            
            # Remove this node from the graph
            for neighbor in graph[node]:
                if neighbor in in_degree:
                    in_degree[neighbor] -= 1
                    if in_degree[neighbor] == 0:
                        queue.append(neighbor)
        
        return result
    
    async def execute_step(self, workflow_id: str, step: WorkflowStep) -> Any:
        """Execute a single workflow step"""
        # Resolve arguments that depend on previous steps
        resolved_args = await self.resolve_arguments(workflow_id, step)
        
        # Execute the tool
        try:
            result = await self.client.call_tool(
                step.server_id, 
                step.tool_name, 
                resolved_args
            )
            return {
                "success": True,
                "result": result,
                "step_id": step.step_id
            }
        except Exception as e:
            return {
                "success": False,
                "error": str(e),
                "step_id": step.step_id
            }
    
    async def resolve_arguments(self, workflow_id: str, step: WorkflowStep) -> Dict:
        """Resolve step arguments, replacing placeholders with previous results"""
        resolved = {}
        
        for key, value in step.arguments.items():
            if isinstance(value, str) and value.startswith("${"):
                # Extract reference to previous step result
                # Format: ${step_id.field_path}
                ref = value[2:-1]  # Remove ${ and }
                resolved[key] = self.resolve_reference(workflow_id, ref)
            else:
                resolved[key] = value
        
        return resolved
    
    def resolve_reference(self, workflow_id: str, reference: str) -> Any:
        """Resolve a reference to a previous step's result"""
        parts = reference.split('.')
        step_id = parts[0]
        
        if step_id not in self.step_results[workflow_id]:
            raise ValueError(f"Step {step_id} not found in workflow results")
        
        result = self.step_results[workflow_id][step_id]
        
        # Navigate through the result using the field path
        for part in parts[1:]:
            if isinstance(result, dict):
                result = result.get(part)
            else:
                raise ValueError(f"Cannot access field {part} in non-dict result")
        
        return result

# Example: Research and Analysis Workflow
async def research_workflow_example():
    """Example of a complex research workflow"""
    
    client = MCPClient("research-client", "1.0.0")
    orchestrator = MCPOrchestrator(client)
    
    # Connect to different MCP servers
    search_server = await client.connect_to_server("http://search-service:8080")
    analysis_server = await client.connect_to_server("http://analysis-service:8080")
    summary_server = await client.connect_to_server("http://summary-service:8080")
    
    # Define workflow steps
    workflow_steps = [
        WorkflowStep(
            step_id="search",
            tool_name="web_search",
            server_id=search_server,
            arguments={
                "query": "artificial intelligence latest developments",
                "max_results": 10
            }
        ),
        WorkflowStep(
            step_id="extract",
            tool_name="extract_key_points",
            server_id=analysis_server,
            arguments={
                "text": "${search.result}",  # Reference to search results
                "max_points": 5
            },
            depends_on=["search"]
        ),
        WorkflowStep(
            step_id="sentiment",
            tool_name="analyze_sentiment",
            server_id=analysis_server,
            arguments={
                "text": "${search.result}"
            },
            depends_on=["search"]
        ),
        WorkflowStep(
            step_id="summarize",
            tool_name="create_summary",
            server_id=summary_server,
            arguments={
                "key_points": "${extract.result}",
                "sentiment": "${sentiment.result}",
                "length": "detailed"
            },
            depends_on=["extract", "sentiment"]
        )
    ]
    
    # Execute the workflow
    results = await orchestrator.execute_workflow("research_001", workflow_steps)
    
    print("Workflow Results:")
    for step_id, result in results.items():
        print(f"{step_id}: {result['success']}")
        if result['success']:
            print(f"  Result: {result['result'][:100]}...")
        else:
            print(f"  Error: {result['error']}")

# Run the workflow example
# asyncio.run(research_workflow_example())

2. Dynamic Tool Discovery and Load Balancing

MCP enables agents to discover tools dynamically and distribute load across multiple servers:

Copy
class MCPLoadBalancer:
    """Load balancer for MCP tool calls"""
    
    def __init__(self, client: MCPClient):
        self.client = client
        self.server_health = {}
        self.tool_mappings = {}  # tool_name -> [server_ids]
    
    async def discover_tools(self):
        """Discover tools across all connected servers"""
        for server_id in self.client.servers:
            try:
                tools = await self.client.list_tools(server_id)
                
                for tool in tools:
                    tool_name = tool["name"]
                    if tool_name not in self.tool_mappings:
                        self.tool_mappings[tool_name] = []
                    
                    if server_id not in self.tool_mappings[tool_name]:
                        self.tool_mappings[tool_name].append(server_id)
                
                self.server_health[server_id] = "healthy"
                
            except Exception as e:
                self.server_health[server_id] = "unhealthy"
                print(f"Server {server_id} health check failed: {e}")
    
    async def call_tool_with_balancing(self, tool_name: str, arguments: Dict) -> Any:
        """Call a tool with load balancing and failover"""
        if tool_name not in self.tool_mappings:
            raise Exception(f"Tool {tool_name} not available")
        
        available_servers = [
            server_id for server_id in self.tool_mappings[tool_name]
            if self.server_health.get(server_id) == "healthy"
        ]
        
        if not available_servers:
            raise Exception(f"No healthy servers available for tool {tool_name}")
        
        # Simple round-robin selection (could be enhanced with actual load metrics)
        selected_server = available_servers[0]
        
        try:
            return await self.client.call_tool(selected_server, tool_name, arguments)
        except Exception as e:
            # Mark server as unhealthy and try another
            self.server_health[selected_server] = "unhealthy"
            
            if len(available_servers) > 1:
                return await self.call_tool_with_balancing(tool_name, arguments)
            else:
                raise e

3. Security and Authorization

MCP provides built-in security mechanisms:

Copy
class SecureMCPServer(MCPServer):
    """MCP Server with enhanced security"""
    
    def __init__(self, name: str, version: str, auth_provider):
        super().__init__(name, version)
        self.auth_provider = auth_provider
        self.authorized_clients = set()
        self.tool_permissions = {}  # tool_name -> required_permissions
    
    async def handle_auth_request(self, params: Dict) -> Dict:
        """Handle client authentication"""
        token = params.get("token")
        client_id = params.get("client_id")
        
        if await self.auth_provider.validate_token(token, client_id):
            self.authorized_clients.add(client_id)
            permissions = await self.auth_provider.get_permissions(client_id)
            return {
                "success": True,
                "permissions": permissions
            }
        else:
            return {
                "success": False,
                "error": "Authentication failed"
            }
    
    async def handle_tools_call(self, params: Dict, client_id: str) -> Dict:
        """Handle tool call with permission checking"""
        tool_name = params["name"]
        
        # Check if client is authorized
        if client_id not in self.authorized_clients:
            raise Exception("Client not authenticated")
        
        # Check tool-specific permissions
        required_perms = self.tool_permissions.get(tool_name, [])
        client_perms = await self.auth_provider.get_permissions(client_id)
        
        if not all(perm in client_perms for perm in required_perms):
            raise Exception(f"Insufficient permissions for tool {tool_name}")
        
        # Execute the tool
        return await super().handle_tools_call(params)
    
    def register_tool_with_permissions(self, tool_spec: Dict, required_permissions: List[str]):
        """Register a tool with specific permission requirements"""
        self.register_tool(tool_spec)
        self.tool_permissions[tool_spec["name"]] = required_permissions

MCP in Practice: Real-World Examples

Example 1: Multi-Modal Research Assistant

Copy
class ResearchAssistantMCP:
    """Research assistant using multiple MCP services"""
    
    def __init__(self):
        self.client = MCPClient("research-assistant", "2.0.0")
        self.orchestrator = MCPOrchestrator(self.client)
        self.servers = {}
    
    async def setup_services(self):
        """Connect to various MCP services"""
        self.servers = {
            "search": await self.client.connect_to_server("http://search-api:8080"),
            "scholar": await self.client.connect_to_server("http://scholar-api:8080"),
            "analysis": await self.client.connect_to_server("http://analysis-api:8080"),
            "vision": await self.client.connect_to_server("http://vision-api:8080"),
            "synthesis": await self.client.connect_to_server("http://synthesis-api:8080")
        }
    
    async def research_topic(self, topic: str, include_images: bool = True) -> Dict:
        """Conduct comprehensive research on a topic"""
        workflow_steps = [
            # Search for text information
            WorkflowStep(
                step_id="web_search",
                tool_name="search_web",
                server_id=self.servers["search"],
                arguments={"query": topic, "type": "comprehensive"}
            ),
            
            # Search academic papers
            WorkflowStep(
                step_id="scholar_search",
                tool_name="search_papers",
                server_id=self.servers["scholar"],
                arguments={"query": topic, "max_papers": 5}
            ),
            
            # Analyze sentiment and key themes
            WorkflowStep(
                step_id="analyze_text",
                tool_name="analyze_themes",
                server_id=self.servers["analysis"],
                arguments={
                    "texts": ["${web_search.result}", "${scholar_search.result}"]
                },
                depends_on=["web_search", "scholar_search"]
            )
        ]
        
        # Add image search if requested
        if include_images:
            workflow_steps.extend([
                WorkflowStep(
                    step_id="image_search",
                    tool_name="search_images",
                    server_id=self.servers["search"],
                    arguments={"query": topic, "count": 10}
                ),
                
                WorkflowStep(
                    step_id="analyze_images",
                    tool_name="analyze_image_content",
                    server_id=self.servers["vision"],
                    arguments={"image_urls": "${image_search.result}"},
                    depends_on=["image_search"]
                )
            ])
        
        # Final synthesis step
        synthesis_args = {
            "text_analysis": "${analyze_text.result}",
            "topic": topic
        }
        
        if include_images:
            synthesis_args["image_analysis"] = "${analyze_images.result}"
            synthesis_step_deps = ["analyze_text", "analyze_images"]
        else:
            synthesis_step_deps = ["analyze_text"]
        
        workflow_steps.append(
            WorkflowStep(
                step_id="synthesize",
                tool_name="create_research_report",
                server_id=self.servers["synthesis"],
                arguments=synthesis_args,
                depends_on=synthesis_step_deps
            )
        )
        
        # Execute the workflow
        results = await self.orchestrator.execute_workflow(f"research_{topic}", workflow_steps)
        
        return results["synthesize"]["result"]

Connections to Previous Concepts

Building on Tool Integration Fundamentals

MCP extends the basic tool integration concepts we learned:

From Tool Integration Fundamentals:

• Function Calling: MCP standardizes function calling across platforms
• Error Handling: MCP provides consistent error reporting mechanisms
• Security: MCP builds in authentication and authorization
• Discovery: MCP enables dynamic tool discovery

Enhanced Capabilities:

• Interoperability: Tools work across different AI frameworks
• Scalability: Distributed tool execution across multiple servers
• Reliability: Built-in failover and load balancing
• Governance: Centralized tool management and monitoring

Practice Exercises

Exercise 1: MCP Server Implementation

Build an MCP server that provides:

1. File system operations (read, write, list directories)
2. Text processing tools (summarize, translate, analyze)
3. Web scraping capabilities
4. Proper authentication and authorization

Ensure your server follows MCP protocol specifications.

Exercise 2: Workflow Orchestration

Create a workflow that:

1. Searches for recent news on a topic
2. Analyzes sentiment across multiple sources
3. Generates a trend analysis
4. Creates visualizations of the findings
5. Compiles everything into a report

Use at least 3 different MCP servers and handle dependencies properly.

Exercise 3: Load Balancing and Failover

Implement an MCP client that:

1. Discovers identical tools across multiple servers
2. Distributes load based on server response times
3. Automatically fails over when servers become unavailable
4. Monitors server health and performance
5. Provides detailed metrics and logging

Exercise 4: Security Analysis

Review the security model of MCP and:

1. Identify potential vulnerabilities
2. Design a secure authentication system
3. Implement rate limiting and access controls
4. Create audit logging for tool usage
5. Propose security best practices for MCP deployments

Looking Ahead

In our next lesson, we'll explore Planning Algorithms and Goal Decomposition. We'll learn how agents can:

• Break down complex goals into manageable subtasks
• Use search algorithms for planning (BFS, DFS, A*)
• Implement hierarchical planning strategies
• Handle planning under uncertainty and changing conditions

The standardized tool ecosystem we've built with MCP will provide the foundation for sophisticated planning systems that can orchestrate complex multi-step procedures.

Additional Resources

• Model Context Protocol Specification
• MCP Python SDK
• MCP TypeScript SDK
• Anthropic MCP Documentation
• JSON-RPC 2.0 Specification

MCP Architecture Visualization

The Model Context Protocol establishes a standardized way for AI agents to securely access tools and resources:

Protocol Flow Visualization

MCP vs Traditional Integration