Ich habe versucht, Langchain mit MCP-Servern zu verwenden, hier sind die Schritte:

Das Model Context Protocol (MCP) ist ein Open-Source-Protokoll, das von Anthropic entwickelt wurde und eine grundlegende Herausforderung in Anwendungen für Large Language Models (LLMs) angeht: ihre Isolation von externen Datenquellen und Tools. Dieses umfassende Tutorial führt Sie durch die Implementierung von MCP mit LangChain und vermittelt Ihnen das Wissen, um anspruchsvolle Anwendungen zu erstellen, die beide Technologien effektiv nutzen.

💡

Bei der Implementierung von Testing für API-basierte Anwendungen greifen Entwickler und Tester zunehmend auf spezialisierte Tools wie Apidog zurück, eine umfassende Postman-Alternative, die den API-Entwicklungslebenszyklus rationalisiert.

button

Verständnis von MCP und seinem Zweck

Das Model Context Protocol zielt darauf ab, die Art und Weise zu standardisieren, wie LLM-basierte Anwendungen eine Verbindung zu verschiedenen externen Systemen herstellen. Stellen Sie sich MCP als "USB-C für KI" vor – eine universelle Schnittstelle, die einen nahtlosen, sicheren und skalierbaren Datenaustausch zwischen LLMs/KI-Agenten und externen Ressourcen ermöglicht.

MCP verwendet eine Client-Server-Architektur:

MCP Hosts: KI-Anwendungen, die auf externe Daten zugreifen müssen
MCP Servers: Daten- oder Toolanbieter, die Informationen für die Hosts bereitstellen

Das Protokoll ermöglicht eine klare Trennung der Zuständigkeiten und ermöglicht es Entwicklern, modulare, wiederverwendbare Konnektoren zu erstellen und gleichzeitig die Sicherheit durch detaillierte Berechtigungskontrollen zu gewährleisten.

Technische Architektur

Die Architektur von MCP besteht aus drei Hauptkomponenten:

Server: Der MCP-Server stellt Tools und Datenquellen über eine standardisierte API bereit
Client: Die Client-Anwendung kommuniziert mit dem Server, um auf Tools und Daten zuzugreifen
Adapter: LangChain bietet Adapter, die die Integration zwischen MCP-Servern und LLM-Anwendungen vereinfachen

Der Kommunikationsfluss folgt diesem Muster:

LangChain-Anwendung fordert Daten/Tool-Ausführung an
MCP-Adapter wandelt die Anfrage in das MCP-Protokollformat um
Server verarbeitet die Anfrage und gibt Ergebnisse zurück
Adapter wandelt die Antwort zurück in ein von LangChain verwendbares Format

Voraussetzungen

Bevor wir beginnen, stellen Sie sicher, dass Sie Folgendes haben:

Python 3.8+ installiert
OpenAI API-Schlüssel (für die Verwendung von GPT-Modellen mit LangChain)
Grundkenntnisse der LangChain-Konzepte
Terminalzugriff (Beispiele werden unter macOS gezeigt)

Einrichten der Umgebung

Erstellen wir zunächst unsere Entwicklungsumgebung und konfigurieren sie:

# Erstellen Sie eine virtuelle Umgebung
python3 -m venv MCP_Demo

# Aktivieren Sie die virtuelle Umgebung
source MCP_Demo/bin/activate

# Installieren Sie die erforderlichen Pakete
pip install langchain-mcp-adapters
pip install langchain-openai

# Legen Sie Ihren OpenAI API-Schlüssel fest
export OPENAI_API_KEY=your_api_key

Erstellen eines einfachen MCP-Servers

Wir beginnen mit dem Erstellen eines einfachen MCP-Servers, der mathematische Operationen bereitstellt. Erstellen Sie eine Datei mit dem Namen math_server.py:

# math_server.py
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("Math")

@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two numbers"""
    return a + b

@mcp.tool()
def multiply(a: int, b: int) -> int:
    """Multiply two numbers"""
    return a * b

if __name__ == "__main__":
    mcp.run(transport="stdio")

Dieser Server stellt zwei mathematische Tools bereit: add und multiply. Die Klasse FastMCP vereinfacht die Servererstellung und behandelt Protokolldetails automatisch. Jede mit @mcp.tool() dekorierte Funktion wird Clients zur Verfügung gestellt, wobei die Dokumentation aus Docstrings abgeleitet wird.

Implementieren des LangChain-Clients

Erstellen Sie als Nächstes einen LangChain-Client, um mit dem MCP-Server zu interagieren. Speichern Sie dies als client.py:

# Create server parameters for stdio connection
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from langchain_mcp_adapters.tools import load_mcp_tools
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
import asyncio

model = ChatOpenAI(model="gpt-4o")

# Configure server parameters
server_params = StdioServerParameters(
    command="python",
    # Specify the path to your server file
    args=["math_server.py"],
)

async def run_agent():
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # Initialize the connection
            await session.initialize()
            
            # Load MCP tools into LangChain format
            tools = await load_mcp_tools(session)
            
            # Create and run the agent
            agent = create_react_agent(model, tools)
            agent_response = await agent.ainvoke({"messages": "what's (3 + 5) x 12?"})
            return agent_response

# Run the async function
if __name__ == "__main__":
    result = asyncio.run(run_agent())
    print(result)

Dieser Client stellt eine Verbindung zum MCP-Server her, lädt die verfügbaren Tools und erstellt einen LangChain-Agenten, der diese Tools zur Lösung von Problemen verwenden kann.

Ausführen des Beispiels

So führen Sie dieses Beispiel aus:

Starten Sie den MCP-Server in einem Terminal-Tab:

python3 math_server.py

Führen Sie im anderen Terminal-Tab den Client aus:

python3 client.py

Der Client ruft den LangChain-Agenten auf, der Folgendes tut:

Analysiert die Frage "(3 + 5) x 12"
Ruft das Tool add mit den Argumenten 3 und 5 auf
Erkennt das Ergebnis 8
Ruft das Tool multiply mit den Argumenten 8 und 12 auf
Gibt die endgültige Antwort zurück: 96

Erweiterte MCP-Server-Implementierung

Erweitern wir unsere Implementierung, um einen anspruchsvolleren MCP-Server zu erstellen, der Datenbankzugriff bietet. Dieses Beispiel zeigt, wie man Konnektoren zu externen Systemen erstellt:

# db_server.py
from mcp.server.fastmcp import FastMCP
import sqlite3
from typing import List, Dict, Any

class DatabaseConnector:
    def __init__(self, db_path):
        self.conn = sqlite3.connect(db_path)
        self.cursor = self.conn.cursor()
    
    def execute_query(self, query: str) -> List[Dict[str, Any]]:
        self.cursor.execute(query)
        columns = [desc[0] for desc in self.cursor.description]
        results = []
        for row in self.cursor.fetchall():
            results.append({columns[i]: row[i] for i in range(len(columns))})
        return results

mcp = FastMCP("DatabaseTools")
db_connector = DatabaseConnector("example.db")

@mcp.tool()
def run_sql_query(query: str) -> List[Dict[str, Any]]:
    """Execute an SQL query on the database and return results"""
    try:
        return db_connector.execute_query(query)
    except Exception as e:
        return {"error": str(e)}

if __name__ == "__main__":
    mcp.run(transport="stdio")

Integrieren mehrerer MCP-Server mit LangChain

Für komplexere Anwendungen müssen Sie möglicherweise mehrere MCP-Server integrieren. So erstellen Sie einen Client, der eine Verbindung zu mehreren Servern herstellt:

# multi_server_client.py
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from langchain_mcp_adapters.tools import load_mcp_tools
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
import asyncio
from typing import List, Dict

# Define our server configurations
servers = [
    {
        "name": "math",
        "params": StdioServerParameters(
            command="python", 
            args=["math_server.py"]
        )
    },
    {
        "name": "database",
        "params": StdioServerParameters(
            command="python", 
            args=["db_server.py"]
        )
    }
]

async def connect_to_server(server_config):
    """Connect to a single MCP server and load its tools"""
    name = server_config["name"]
    params = server_config["params"]
    
    read, write = await stdio_client(params).__aenter__()
    session = ClientSession(read, write)
    await session.__aenter__()
    await session.initialize()
    
    tools = await load_mcp_tools(session)
    return {
        "name": name,
        "session": session,
        "tools": tools,
        "cleanup": lambda: asyncio.gather(
            session.__aexit__(None, None, None),
            stdio_client(params).__aexit__(None, None, None)
        )
    }

async def run_multi_server_agent():
    # Connect to all servers
    connections = await asyncio.gather(
        *[connect_to_server(server) for server in servers]
    )
    
    try:
        # Collect all tools from all servers
        all_tools = []
        for connection in connections:
            all_tools.extend(connection["tools"])
        
        # Create the agent with all tools
        model = ChatOpenAI(model="gpt-4o")
        agent = create_react_agent(model, all_tools)
        
        # Run the agent with a complex query that might use multiple servers
        response = await agent.ainvoke({
            "messages": "Find the customers who've spent more than the average order value and calculate their total spend."
        })
        
        return response
    
    finally:
        # Clean up all connections
        for connection in connections:
            await connection["cleanup"]()

# Run the multi-server agent
if __name__ == "__main__":
    result = asyncio.run(run_multi_server_agent())
    print(result)

Fehlerbehandlung und Fallback-Strategien

Robuste MCP-Implementierungen sollten eine Fehlerbehandlung umfassen. Hier ist eine erweiterte Version des Clients, die Fehlerbehandlung und Fallback-Strategien demonstriert:

# robust_client.py
async def run_agent_with_fallbacks():
    try:
        # Attempt primary connection
        async with stdio_client(server_params) as (read, write):
            async with ClientSession(read, write) as session:
                try:
                    await session.initialize()
                    tools = await load_mcp_tools(session)
                    agent = create_react_agent(model, tools)
                    return await agent.ainvoke({"messages": "what's (3 + 5) x 12?"})
                except Exception as e:
                    print(f"Error using MCP tools: {e}")
                    # Fallback to direct model call without tools
                    return await model.ainvoke([
                        HumanMessage(content="what's (3 + 5) x 12?")
                    ])
    except Exception as connection_error:
        print(f"Connection error: {connection_error}")
        # Ultimate fallback
        return {"error": "Could not establish connection to MCP server"}

Sicherheitsüberlegungen

Berücksichtigen Sie bei der Implementierung von MCP mit LangChain diese bewährten Sicherheitspraktiken:

Eingabevalidierung: Validieren Sie immer Eingaben für MCP-Tools, um Injektionsangriffe zu verhindern
Tool-Berechtigungen: Implementieren Sie detaillierte Berechtigungen für jedes Tool
Ratenbegrenzung: Wenden Sie Ratenbegrenzungen an, um den Missbrauch von Tools zu verhindern
Authentifizierung: Implementieren Sie eine ordnungsgemäße Authentifizierung zwischen Clients und Servern

Hier ist ein Beispiel für die Implementierung von Tool-Berechtigungen:

from mcp.server.fastmcp import FastMCP, Permission

mcp = FastMCP("SecureTools")

# Define permission levels
READ_PERMISSION = Permission("read", "Can read data")
WRITE_PERMISSION = Permission("write", "Can modify data")

@mcp.tool(permissions=[READ_PERMISSION])
def get_data(key: str) -> str:
    """Get data by key (requires read permission)"""
    # Implementation...
    return f"Data for {key}"

@mcp.tool(permissions=[WRITE_PERMISSION])
def update_data(key: str, value: str) -> bool:
    """Update data (requires write permission)"""
    # Implementation...
    return True

Leistungsoptimierung

Berücksichtigen Sie für Produktionsbereitstellungen diese Leistungsoptimierungen:

Verbindungspooling: Verwenden Sie MCP-Verbindungen wieder, anstatt für jede Anfrage neue zu erstellen
Stapelverarbeitung: Gruppieren Sie nach Möglichkeit mehrere Toolaufrufe
Asynchrone Verarbeitung: Verwenden Sie asyncio, um mehrere Anfragen gleichzeitig zu bearbeiten

Beispiel für Verbindungspooling:

class MCPConnectionPool:
    def __init__(self, max_connections=10):
        self.available_connections = asyncio.Queue(max_connections)
        self.max_connections = max_connections
        self.current_connections = 0
    
    async def initialize(self):
        # Pre-create some connections
        for _ in range(3):  # Start with 3 connections
            await self._create_connection()
    
    async def _create_connection(self):
        if self.current_connections >= self.max_connections:
            raise Exception("Maximum connections reached")
        
        read, write = await stdio_client(server_params).__aenter__()
        session = await ClientSession(read, write).__aenter__()
        await session.initialize()
        
        self.current_connections += 1
        await self.available_connections.put(session)
    
    async def get_connection(self):
        if self.available_connections.empty() and self.current_connections < self.max_connections:
            await self._create_connection()
        
        return await self.available_connections.get()
    
    async def release_connection(self, connection):
        await self.available_connections.put(connection)

Testen von MCP-Implementierungen

Gründliches Testen ist für zuverlässige MCP-Implementierungen unerlässlich. Hier ist ein Testansatz mit pytest:

# test_mcp.py
import pytest
import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from langchain_mcp_adapters.tools import load_mcp_tools

@pytest.fixture
async def mcp_session():
    server_params = StdioServerParameters(
        command="python",
        args=["math_server.py"],
    )
    
    read, write = await stdio_client(server_params).__aenter__()
    session = ClientSession(read, write)
    await session.__aenter__()
    await session.initialize()
    
    yield session
    
    await session.__aexit__(None, None, None)
    await stdio_client(server_params).__aexit__(None, None, None)

@pytest.mark.asyncio
async def test_add_tool(mcp_session):
    tools = await load_mcp_tools(mcp_session)
    add_tool = next(tool for tool in tools if tool.name == "add")
    
    result = await add_tool.invoke({"a": 5, "b": 7})
    assert result == 12

@pytest.mark.asyncio
async def test_multiply_tool(mcp_session):
    tools = await load_mcp_tools(mcp_session)
    multiply_tool = next(tool for tool in tools if tool.name == "multiply")
    
    result = await multiply_tool.invoke({"a": 6, "b": 8})
    assert result == 48

Fazit

Das Model Context Protocol bietet einen leistungsstarken Rahmen für die Verbindung von LangChain-Anwendungen mit externen Tools und Datenquellen. Durch die Standardisierung dieser Verbindungen ermöglicht MCP Entwicklern, anspruchsvolle KI-Agenten zu erstellen, die nahtlos mit ihrer Umgebung interagieren können.

Die Kombination aus den Agentenfunktionen von LangChain und der Konnektivität von MCP schafft eine Grundlage für den Aufbau wirklich leistungsstarker, kontextsensitiver Anwendungen. Da das MCP-Ökosystem weiter wächst, können wir erwarten, dass weitere vorgefertigte Server und Tools entstehen, die den Entwicklungsprozess weiter vereinfachen.

Dieses Tutorial hat die grundlegenden Konzepte und Implementierungsdetails der Verwendung von MCP mit LangChain behandelt, von der grundlegenden Einrichtung bis hin zu erweiterten Mustern wie Verbindungspooling und Fehlerbehandlung. Wenn Sie diese Praktiken befolgen, können Sie robuste, produktionsbereite Anwendungen erstellen, die das Beste aus beiden Technologien nutzen.

Für weitere Erkundungen sollten Sie das wachsende Ökosystem von MCP-Servern untersuchen, das auf GitHub verfügbar ist, oder Ihre eigenen Server zur Community beitragen. Die Zukunft der KI-Agenten liegt in ihrer Fähigkeit, externe Tools und Daten effektiv zu nutzen, und MCP ist ein wichtiger Schritt, um diese Vision zu verwirklichen.