13. MCP with AI Agents

flowchart TD
    AGENT["🤖 AI Agent\n(Plans, Reasons, Decides)"]
    AGENT -->|"I need to search"| MCP["🔗 MCP Layer\n(Standard Protocol)"]
    MCP --> TOOLS["🔧 Tools\n(search_files, query_db)"]
    MCP --> RESOURCES["📄 Resources\n(file://, docs://)"]
    MCP --> PROMPTS["💬 Prompts\n(code_review, summarize)"]
    TOOLS --> BACKEND["External Systems"]
    RESOURCES --> BACKEND

    style AGENT fill:#3b82f6,color:#fff
    style MCP fill:#22c55e,color:#fff

flowchart TD
    subgraph AGENTS["Agent Frameworks"]
        LG["LangGraph Agent"]
        CW["CrewAI Agent"]
        OA["OpenAI Agent"]
        CD["Claude Desktop"]
    end

    subgraph MCP_LAYER["MCP Integration Layer"]
        MC["MCP Client Manager\n(one client per server)"]
        RR["Router\n(routes tool calls to server)"]
    end

    subgraph SERVERS["MCP Servers"]
        FS["📁 Filesystem MCP"]
        DB["🗄️ Database MCP"]
        GH["🐙 GitHub MCP"]
        SL["💬 Slack MCP"]
    end

    AGENTS -->|"Tool call"| MCP_LAYER
    MCP_LAYER -->|"tools/call"| SERVERS
    SERVERS -->|"Result"| MCP_LAYER
    MCP_LAYER -->|"Formatted result"| AGENTS

    style AGENTS fill:#3b82f6,color:#fff
    style MCP_LAYER fill:#22c55e,color:#fff
    style SERVERS fill:#f59e0b,color:#fff

import asyncio
from typing import Literal
from langgraph.graph import StateGraph, MessagesState, END
from langgraph.prebuilt import ToolNode
from mcp import ClientSession
from mcp.client.stdio import stdio_client, StdioServerParameters

class MCPToolWrapper:
    """Wraps an MCP tool for LangGraph."""

    def __init__(self, session: ClientSession, tool_def):
        self.session = session
        self.name = tool_def.name
        self.description = tool_def.description
        # Convert MCP schema to LangGraph-compatible format
        self.schema = tool_def.inputSchema

    async def __call__(self, **kwargs):
        result = await self.session.call_tool(self.name, kwargs)
        return result.content[0].text if result.content else ""

async def build_mcp_agent():
    # Connect to MCP server
    server_params = StdioServerParameters(
        command="python",
        args=["-m", "my_mcp_server"]
    )

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # Discover tools
            tools = await session.list_tools()
            wrapped_tools = [MCPToolWrapper(session, t) for t in tools]

            # Build LangGraph with MCP tools
            tool_node = ToolNode(wrapped_tools)

            # Define the agent workflow
            workflow = StateGraph(MessagesState)

            def call_agent(state):
                # Agent logic — decides which tool to call
                return {"messages": [decide_next_action(state)]}

            def should_continue(state) -> Literal["tools", END]:
                last = state["messages"][-1]
                return "tools" if hasattr(last, "tool_calls") else END

            workflow.add_node("agent", call_agent)
            workflow.add_node("tools", tool_node)
            workflow.add_edge("tools", "agent")
            workflow.add_conditional_edges("agent", should_continue)
            workflow.set_entry_point("agent")

            app = workflow.compile()
            return app

sequenceDiagram
    participant Agent as LangGraph Agent
    participant MCP as MCP Client
    participant Server as MCP Server

    Agent->>MCP: Initialize & discover tools
    MCP->>Server: initialize + tools/list
    Server->>MCP: Tool definitions
    MCP->>Agent: Wrapped tools

    Agent->>Agent: Decides to call search_docs

    Agent->>MCP: search_docs(query="MCP")
    MCP->>Server: tools/call(name, arguments)
    Server->>Server: Execute search
    Server->>MCP: Results
    MCP->>Agent: Formatted result

    Agent->>Agent: Processes result, decides next action

from crewai import Agent, Task, Crew, Process
from mcp import ClientSession
from mcp.client.stdio import stdio_client, StdioServerParameters
from typing import Any
import json

class MCPToolProvider:
    """Provides MCP server tools as CrewAI-compatible tools."""

    def __init__(self, server_command: str, server_args: list[str]):
        self.server_command = server_command
        self.server_args = server_args
        self.session = None
        self.read = None
        self.write = None

    async def connect(self):
        params = StdioServerParameters(
            command=self.server_command,
            args=self.server_args
        )
        self.read, self.write = await stdio_client(params).__aenter__()
        self.session = await ClientSession(self.read, self.write).__aenter__()
        await self.session.initialize()

    def create_tool(self, tool_def):
        """Create a CrewAI-compatible tool from MCP tool definition."""
        tool_name = tool_def.name
        tool_description = tool_def.description

        async def tool_function(**kwargs):
            result = await self.session.call_tool(tool_name, kwargs)
            return result.content[0].text if result.content else ""

        tool_function.__name__ = tool_name
        tool_function.__doc__ = tool_description
        return tool_function

# Usage with CrewAI
async def run_crew_with_mcp():
    provider = MCPToolProvider("python", ["-m", "doc_server"])
    await provider.connect()

    # Discover and create tools
    tools = await provider.session.list_tools()
    crew_tools = [provider.create_tool(t) for t in tools]

    # Create CrewAI agent with MCP tools
    researcher = Agent(
        role="Research Specialist",
        goal="Find and analyze documentation",
        tools=crew_tools,
        backstory="Expert at searching documentation"
    )

    task = Task(
        description="Search for MCP documentation and summarize findings",
        agent=researcher
    )

    crew = Crew(
        agents=[researcher],
        tasks=[task],
        process=Process.sequential
    )

    result = crew.kickoff()
    return result

flowchart TD
    subgraph CREW["CrewAI"]
        MANAGER["Manager Agent"]
        RESEARCH["Research Agent\n(MCP Tools)"]
        WRITER["Writer Agent"]
    end

    subgraph MCP["MCP Layer"]
        MCPC["MCP Client"]
    end

    subgraph SERVERS["MCP Servers"]
        DOCS["Doc Search Server"]
        FS["Filesystem Server"]
    end

    MANAGER -->|"Assign task"| RESEARCH
    RESEARCH -->|"search_docs()"| MCPC
    MCPC -->|"Query"| DOCS
    DOCS -->|"Results"| MCPC
    MCPC -->|"Formatted"| RESEARCH
    RESEARCH -->|"Findings"| WRITER
    WRITER -->|"Final output"| MANAGER

    style CREW fill:#3b82f6,color:#fff
    style MCP fill:#22c55e,color:#fff
    style SERVERS fill:#f59e0b,color:#fff

{
  "mcpServers": {
    "filesystem": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-filesystem", "/path/to/allowed/files"]
    },
    "github": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-github"],
      "env": {
        "GITHUB_TOKEN": "gh_..."
      }
    },
    "database": {
      "command": "python",
      "args": ["-m", "my_db_server"],
      "env": {
        "DATABASE_URL": "postgresql://..."
      }
    }
  }
}

flowchart TD
    subgraph CLAUDE["Claude Desktop"]
        CLAUDE_AGENT["Claude AI"]
        MCP_CLIENT["Built-in MCP Client"]
    end

    subgraph SERVERS["MCP Servers"]
        FS["📁 Filesystem"]
        GH["🐙 GitHub"]
        DB["🗄️ Database"]
        CUSTOM["⚙️ Custom Server"]
    end

    CLAUDE_AGENT -->|"User asks to read a file"| MCP_CLIENT
    MCP_CLIENT -->|"STDIO"| FS
    FS -->|"File content"| CLAUDE_AGENT

    CLAUDE_AGENT -->|"User asks about repo"| MCP_CLIENT
    MCP_CLIENT -->|"STDIO"| GH
    GH -->|"Repo data"| CLAUDE_AGENT

    style CLAUDE fill:#3b82f6,color:#fff
    style SERVERS fill:#22c55e,color:#fff

from openai import OpenAI
from mcp import ClientSession
from mcp.client.stdio import stdio_client, StdioServerParameters
import json

class MCPFunctionProvider:
    """Converts MCP tools to OpenAI function definitions."""

    def __init__(self):
        self.session = None

    async def connect(self, command: str, args: list[str]):
        params = StdioServerParameters(command=command, args=args)
        read, write = await stdio_client(params).__aenter__()
        self.session = await ClientSession(read, write).__aenter__()
        await self.session.initialize()

    def to_openai_functions(self, mcp_tools):
        """Convert MCP tool definitions to OpenAI function format."""
        functions = []
        for tool in mcp_tools:
            functions.append({
                "type": "function",
                "function": {
                    "name": tool.name,
                    "description": tool.description,
                    "parameters": tool.inputSchema
                }
            })
        return functions

    async def execute_function(self, name: str, arguments: dict):
        result = await self.session.call_tool(name, arguments)
        return result.content[0].text if result.content else ""

# Usage with OpenAI
async def run_openai_agent():
    provider = MCPFunctionProvider()
    await provider.connect("python", ["-m", "doc_server"])

    tools = await provider.session.list_tools()
    functions = provider.to_openai_functions(tools)

    client = OpenAI()

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": "Search for MCP documentation"}],
        tools=functions,
        tool_choice="auto"
    )

    # Handle function calls
    for choice in response.choices:
        if choice.message.tool_calls:
            for tool_call in choice.message.tool_calls:
                name = tool_call.function.name
                args = json.loads(tool_call.function.arguments)
                result = await provider.execute_function(name, args)
                print(f"Tool result: {result}")

flowchart TD
    subgraph AGENTS["Multi-Agent System"]
        COORD["🤖 Coordinator Agent"]
        RESEARCH["🔍 Research Agent"]
        CODE["💻 Coding Agent"]
        REVIEW["✅ Review Agent"]
    end

    subgraph MCP["MCP Layer"]
        ROUTER["MCP Router\n(routes to correct server)"]
        CACHE["Tool Cache\n(cached schemas)"]
    end

    subgraph SERVERS["MCP Servers"]
        WEB["🌐 Web Search"]
        FS["📁 Filesystem"]
        GH["🐙 GitHub"]
        SQL["🗄️ Database"]
    end

    COORD -->|"Assigns tasks"| RESEARCH
    COORD -->|"Assigns tasks"| CODE
    COORD -->|"Assigns tasks"| REVIEW

    RESEARCH -->|"search_web()"| ROUTER
    CODE -->|"read_file()"| ROUTER
    CODE -->|"create_pr()"| ROUTER
    REVIEW -->|"read_file()"| ROUTER

    ROUTER --> WEB
    ROUTER --> FS
    ROUTER --> GH
    ROUTER --> SQL

    style AGENTS fill:#3b82f6,color:#fff
    style MCP fill:#22c55e,color:#fff
    style SERVERS fill:#f59e0b,color:#fff