Intenté usar Langchain con servidores MCP, aquí están los pasos:

El Protocolo de Contexto del Modelo (MCP) es un protocolo de código abierto desarrollado por Anthropic que aborda un desafío fundamental en las aplicaciones de Modelos de Lenguaje Grande (LLM): su aislamiento de fuentes de datos y herramientas externas. Este tutorial completo le guiará a través de la implementación de MCP con LangChain, proporcionándole el conocimiento para crear aplicaciones sofisticadas que aprovechen ambas tecnologías de manera efectiva.

💡

Al implementar pruebas para aplicaciones basadas en API, los desarrolladores y testers recurren cada vez más a herramientas especializadas como Apidog, una alternativa integral a Postman que agiliza el ciclo de vida del desarrollo de API.

button

Comprendiendo MCP y su Propósito

El Protocolo de Contexto del Modelo tiene como objetivo estandarizar cómo las aplicaciones basadas en LLM se conectan a diversos sistemas externos. Piense en MCP como el "USB-C para la IA": una interfaz universal que permite el intercambio de datos sin problemas, seguro y escalable entre LLM/agentes de IA y recursos externos.

MCP emplea una arquitectura cliente-servidor:

Hosts MCP: Aplicaciones de IA que necesitan acceder a datos externos
Servidores MCP: Proveedores de datos o herramientas que suministran información a los hosts

El protocolo facilita una clara separación de responsabilidades, permitiendo a los desarrolladores crear conectores modulares y reutilizables, manteniendo al mismo tiempo una seguridad robusta a través de controles de permisos granulares.

Arquitectura Técnica

La arquitectura de MCP consta de tres componentes principales:

Servidor: El servidor MCP expone herramientas y fuentes de datos a través de una API estandarizada
Cliente: La aplicación cliente se comunica con el servidor para acceder a herramientas y datos
Adaptador: LangChain proporciona adaptadores que simplifican la integración entre los servidores MCP y las aplicaciones LLM

El flujo de comunicación sigue este patrón:

La aplicación LangChain solicita la ejecución de datos/herramientas
El adaptador MCP transforma la solicitud en el formato del protocolo MCP
El servidor procesa la solicitud y devuelve los resultados
El adaptador transforma la respuesta de nuevo a un formato utilizable por LangChain

Prerrequisitos

Antes de comenzar, asegúrese de tener lo siguiente:

Python 3.8+ instalado
Clave de la API de OpenAI (para usar modelos GPT con LangChain)
Familiaridad básica con los conceptos de LangChain
Acceso al terminal (ejemplos mostrados en macOS)

Configurando el Entorno

Primero, vamos a crear y configurar nuestro entorno de desarrollo:

# Create a virtual environment
python3 -m venv MCP_Demo

# Activate the virtual environment
source MCP_Demo/bin/activate

# Install required packages
pip install langchain-mcp-adapters
pip install langchain-openai

# Set your OpenAI API key
export OPENAI_API_KEY=your_api_key

Creando un Servidor MCP Simple

Comenzaremos construyendo un servidor MCP básico que proporcione operaciones matemáticas. Cree un archivo llamado 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")

Este servidor expone dos herramientas matemáticas: add y multiply. La clase FastMCP simplifica la creación del servidor, manejando los detalles del protocolo automáticamente. Cada función decorada con @mcp.tool() se vuelve disponible para los clientes, con documentación derivada de docstrings.

Implementando el Cliente LangChain

A continuación, cree un cliente LangChain para interactuar con el servidor MCP. Guarde esto como 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)

Este cliente establece una conexión con el servidor MCP, carga las herramientas disponibles y crea un agente LangChain que puede usar estas herramientas para resolver problemas.

Ejecutando el Ejemplo

Para ejecutar este ejemplo:

Inicie el servidor MCP en una pestaña del terminal:

python3 math_server.py

En otra pestaña del terminal, ejecute el cliente:

python3 client.py

El cliente invocará al agente LangChain, que:

Analizará la pregunta "(3 + 5) x 12"
Llamará a la herramienta add con los argumentos 3 y 5
Obtendrá el resultado 8
Llamará a la herramienta multiply con los argumentos 8 y 12
Devolverá la respuesta final: 96

Implementación Avanzada del Servidor MCP

Ampliemos nuestra implementación para crear un servidor MCP más sofisticado que proporcione acceso a la base de datos. Este ejemplo demuestra cómo construir conectores a sistemas externos:

# 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")

Integrando Múltiples Servidores MCP con LangChain

Para aplicaciones más complejas, es posible que necesite integrar varios servidores MCP. Aquí se muestra cómo crear un cliente que se conecta a varios servidores:

# 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)

Manejo de Errores y Estrategias de Respaldo

Las implementaciones robustas de MCP deben incluir el manejo de errores. Aquí hay una versión mejorada del cliente que demuestra el manejo de errores y las estrategias de respaldo:

# 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"}

Consideraciones de Seguridad

Al implementar MCP con LangChain, considere estas prácticas recomendadas de seguridad:

Validación de Entrada: Siempre valide las entradas a las herramientas MCP para evitar ataques de inyección
Permisos de Herramientas: Implemente permisos detallados para cada herramienta
Limitación de Tasa: Aplique límites de tasa para evitar el abuso de herramientas
Autenticación: Implemente la autenticación adecuada entre clientes y servidores

Aquí hay un ejemplo de implementación de permisos de herramientas:

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

Optimización del Rendimiento

Para implementaciones de producción, considere estas optimizaciones de rendimiento:

Agrupación de Conexiones: Reutilice las conexiones MCP en lugar de crear nuevas para cada solicitud
Procesamiento por Lotes: Agrupe varias llamadas a herramientas cuando sea posible
Procesamiento Asíncrono: Use asyncio para manejar múltiples solicitudes simultáneamente

Ejemplo de agrupación de conexiones:

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)

Probando Implementaciones de MCP

Las pruebas exhaustivas son cruciales para implementaciones de MCP confiables. Aquí hay un enfoque de prueba usando 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

Conclusión

El Protocolo de Contexto del Modelo proporciona un marco poderoso para conectar aplicaciones LangChain con herramientas y fuentes de datos externas. Al estandarizar estas conexiones, MCP permite a los desarrolladores crear agentes de IA sofisticados que pueden interactuar sin problemas con su entorno.

La combinación de las capacidades del agente de LangChain con la conectividad de MCP crea una base para construir aplicaciones verdaderamente poderosas y conscientes del contexto. A medida que el ecosistema MCP continúa creciendo, podemos esperar que surjan más servidores y herramientas preconstruidos, simplificando aún más el proceso de desarrollo.

Este tutorial ha cubierto los conceptos fundamentales y los detalles de implementación del uso de MCP con LangChain, desde la configuración básica hasta patrones avanzados como la agrupación de conexiones y el manejo de errores. Siguiendo estas prácticas, puede crear aplicaciones robustas y listas para producción que aprovechen lo mejor de ambas tecnologías.

Para una mayor exploración, considere investigar el creciente ecosistema de servidores MCP disponibles en GitHub, o contribuya con sus propios servidores a la comunidad. El futuro de los agentes de IA radica en su capacidad para aprovechar eficazmente las herramientas y los datos externos, y MCP es un paso significativo para hacer realidad esa visión.