Como Integrar o LangChain com Servidores MCP Usando langchain-mcp-adapters

O Protocolo de Contexto do Modelo (MCP) visa padronizar como modelos de IA interagem com ferramentas e serviços externos. Ele define uma interface comum, permitindo que diferentes modelos e provedores de ferramentas se comuniquem de forma eficaz. No entanto, integrar essas ferramentas compatíveis com o MCP diretamente em frameworks de IA existentes, como o LangChain, requer adaptação.

É aqui que entra a biblioteca langchain-mcp-adapters. Ela atua como uma ponte crucial, traduzindo sem problemas ferramentas MCP em um formato que o LangChain e seu poderoso framework de agentes, LangGraph, podem entender e utilizar. Esta biblioteca fornece um wrapper leve, permitindo que desenvolvedores aproveitem o crescente ecossistema de ferramentas MCP dentro de suas aplicações LangChain.

Os principais recursos incluem:

Conversão de Ferramentas MCP: Converte automaticamente ferramentas MCP em objetos BaseTool compatíveis com LangChain.
Cliente Multi-Servidor: Fornece um cliente robusto (MultiServerMCPClient) capaz de se conectar a vários servidores MCP simultaneamente, agregando ferramentas de várias fontes.
Flexibilidade de Transporte: Suporta transportes de comunicação MCP comuns, como entrada/saída padrão (stdio) e Eventos Enviados pelo Servidor (sse).

Este tutorial irá guiá-lo através da configuração de servidores MCP, conectando-se a eles usando a biblioteca de adaptadores e integrando as ferramentas carregadas em um agente LangGraph.

💡

Quer uma ótima ferramenta de teste de API que gera documentação de API bonita?

Quer uma plataforma integrada Tudo-em-Um para sua equipe de desenvolvedores trabalhar junto com máxima produtividade?

Apidog atende todas as suas demandas e substitui o Postman a um preço muito mais acessível!

button

O que é o Servidor MCP? Como Funciona?

Compreender alguns conceitos centrais é essencial antes de mergulhar nos exemplos:

Servidor MCP:

Um servidor MCP expõe ferramentas (funções) que um modelo de IA pode chamar.
A biblioteca mcp (uma dependência do langchain-mcp-adapters) fornece ferramentas como FastMCP para criar facilmente esses servidores em Python.
As ferramentas são definidas usando o decorador @mcp.tool(), que infere automaticamente o esquema de entrada a partir de dicas de tipo e docstrings.
Os servidores também podem definir prompts usando @mcp.prompt(), fornecendo iniciadores ou instruções conversacionais estruturadas.
Os servidores são executados especificando um mecanismo de transporte (por exemplo, mcp.run(transport="stdio") ou mcp.run(transport="sse")). O stdio executa o servidor como um subprocesso se comunicando via entrada/saída padrão, enquanto sse geralmente executa um simples servidor web para comunicação.

Cliente MCP (langchain-mcp-adapters):

O papel do cliente é conectar-se a um ou mais servidores MCP.
Ele lida com os detalhes do protocolo de comunicação (stdio, sse).
Ele busca a lista de ferramentas disponíveis e suas definições (nome, descrição, esquema de entrada) do(s) servidor(es).
A classe MultiServerMCPClient é a principal forma de gerenciar conexões, especialmente ao lidar com vários servidores de ferramentas.

Conversão de Ferramentas:

As ferramentas MCP têm seu próprio formato de definição. O LangChain utiliza sua estrutura de classe BaseTool.
A biblioteca langchain-mcp-adapters fornece funções como load_mcp_tools (encontrada em langchain_mcp_adapters.tools) que conectam-se a um servidor via uma ClientSession ativa, listam as ferramentas MCP e envolvem cada uma em uma StructuredTool do LangChain.
Esse wrapper lida com a invocação da chamada real da ferramenta MCP (session.call_tool) quando o agente LangChain decide usar a ferramenta e formata corretamente a resposta.

Conversão de Prompt:

Semelhante às ferramentas, os prompts MCP podem ser buscados usando load_mcp_prompt (de langchain_mcp_adapters.prompts).
Esta função recupera a estrutura do prompt do servidor MCP e a converte em uma lista de objetos HumanMessage ou AIMessage do LangChain, adequados para inicializar ou guiar uma conversa.

Instalar o Langchain-mcp-adapter

Primeiro, instale os pacotes necessários:

pip install langchain-mcp-adapters langgraph langchain-openai # Ou a sua integração preferida do LangChain LLM

Você também precisará configurar chaves de API para o provedor de modelo de linguagem escolhido, normalmente definindo variáveis de ambiente:

export OPENAI_API_KEY=<your_openai_api_key>
# ou export ANTHROPIC_API_KEY=<...> etc.

Construir um Servidor MCP Simples com langchain-mcp-adapters

Vamos construir um exemplo simples: um servidor MCP fornecendo funções matemáticas e um agente LangGraph usando essas funções.

Passo 1: Criar o Servidor MCP (math_server.py)

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

# Inicializa o servidor MCP com um nome
mcp = FastMCP("Math")

@mcp.tool()
def add(a: int, b: int) -> int:
    """Adiciona dois números"""
    print(f"Executando add({a}, {b})") # Log do lado do servidor
    return a + b

@mcp.tool()
def multiply(a: int, b: int) -> int:
    """Multiplica dois números"""
    print(f"Executando multiply({a}, {b})") # Log do lado do servidor
    return a * b

# Definição do exemplo de prompt
@mcp.prompt()
def configure_assistant(skills: str) -> list[dict]:
    """Configura o assistente com habilidades especificadas."""
    return [
        {
            "role": "assistant", # Corresponde a AIMessage
            "content": f"Você é um assistente prestativo. Você possui as seguintes habilidades: {skills}. Sempre use apenas uma ferramenta por vez.",
        }
    ]

if __name__ == "__main__":
    # Execute o servidor usando transporte stdio
    print("Iniciando o servidor Math MCP via stdio...")
    mcp.run(transport="stdio")

Salve este código como math_server.py.

Passo 2: Criar o Cliente e o Agente (client_app.py)

import asyncio
import os
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

# --- IMPORTANTE: Atualize este caminho ---
# Obter o caminho absoluto do arquivo math_server.py
current_dir = os.path.dirname(os.path.abspath(__file__))
math_server_script_path = os.path.join(current_dir, "math_server.py")
# ---

async def main():
    model = ChatOpenAI(model="gpt-4o") # Ou seu modelo preferido

    # Configure parâmetros para executar o script math_server.py
    server_params = StdioServerParameters(
        command="python", # O comando para executar
        args=[math_server_script_path], # Argumentos (o caminho do script)
        # cwd=..., env=... # Diretório de trabalho opcional e variáveis de ambiente
    )

    print("Conectando ao servidor MCP...")
    # Estabelece conexão usando o gerenciador de contexto stdio_client
    async with stdio_client(server_params) as (read, write):
        # Cria uma ClientSession usando os streams de leitura/escrita
        async with ClientSession(read, write) as session:
            print("Inicializando sessão...")
            # Handshake com o servidor
            await session.initialize()
            print("Sessão inicializada.")

            print("Carregando ferramentas MCP...")
            # Busca ferramentas MCP e converte-as em ferramentas do LangChain
            tools = await load_mcp_tools(session)
            print(f"Ferramentas carregadas: {[tool.name for tool in tools]}")

            # Cria um agente LangGraph ReAct usando o modelo e as ferramentas carregadas
            agent = create_react_agent(model, tools)

            print("Invocando agente...")
            # Executa o agente
            inputs = {"messages": [("human", "qual é (3 + 5) * 12?")]}
            async for event in agent.astream_events(inputs, version="v1"):
                 print(event) # Eventos de stream para observabilidade

            # Ou obtenha a resposta final diretamente
            # final_response = await agent.ainvoke(inputs)
            # print("Resposta do agente:", final_response['messages'][-1].content)

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

Salve isso como client_app.py no mesmo diretório que math_server.py.

Para Executar:

Execute o script do cliente:

python client_app.py

O script do cliente iniciará automaticamente math_server.py como um subprocesso, conectando-se a ele, carregando as ferramentas add e multiply, e usando o agente LangGraph para resolver o problema matemático chamando essas ferramentas via o servidor MCP. Você verá logs de ambos, o cliente e o servidor.

Conectando a Vários Servidores MCP

Freqüentemente, você desejará combinar ferramentas de diferentes servidores especializados. O MultiServerMCPClient torna isso simples.

Passo 1: Criar Outro Servidor (weather_server.py)

Vamos criar um servidor de clima que funciona usando transporte SSE.

# weather_server.py
from mcp.server.fastmcp import FastMCP
import uvicorn # Necessita: pip install uvicorn

mcp = FastMCP("Weather")

@mcp.tool()
async def get_weather(location: str) -> str:
    """Obtém o clima para a localização."""
    print(f"Executando get_weather({location})")
    # Em uma situação real, isso chamaria uma API de clima
    return f"Está sempre ensolarado em {location}"

if __name__ == "__main__":
    # Execute o servidor usando transporte SSE (exige um servidor ASGI como uvicorn)
    # A biblioteca mcp cria implicitamente um app FastAPI para SSE.
    # Por padrão, ele é executado na porta 8000 no endpoint /sse.
    print("Iniciando o servidor Weather MCP via SSE na porta 8000...")
    # uvicorn.run(mcp.app, host="0.0.0.0", port=8000) # Você pode executar manualmente
    mcp.run(transport="sse", host="0.0.0.0", port=8000) # Ou use o mcp.run para conveniência

Salve isso como weather_server.py.

Passo 2: Atualizar Cliente para Usar MultiServerMCPClient (multi_client_app.py)

import asyncio
import os
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI

# --- IMPORTANTE: Atualize os caminhos ---
current_dir = os.path.dirname(os.path.abspath(__file__))
math_server_script_path = os.path.join(current_dir, "math_server.py")
# O servidor de clima roda separadamente, conecte-se via URL
# ---

async def main():
    model = ChatOpenAI(model="gpt-4o")

    # Defina conexões para múltiplos servidores
    server_connections = {
        "math_service": { # Nome único para esta conexão
            "transport": "stdio",
            "command": "python",
            "args": [math_server_script_path],
            # Adicione outros parâmetros de StdioConnection se necessário (env, cwd, etc.)
        },
        "weather_service": { # Nome único para esta conexão
            "transport": "sse",
            "url": "http://localhost:8000/sse", # URL onde o weather_server está rodando
            # Adicione outros parâmetros de SSEConnection se necessário (headers, timeout, etc.)
        }
    }

    print("Conectando a vários servidores MCP...")
    # Use o gerenciador de contexto MultiServerMCPClient
    async with MultiServerMCPClient(server_connections) as client:
        print("Conexões estabelecidas.")

        # Obtenha *todas* as ferramentas de *todos* os servidores conectados
        all_tools = client.get_tools()
        print(f"Ferramentas carregadas: {[tool.name for tool in all_tools]}")

        # Crie um agente com a lista combinada de ferramentas
        agent = create_react_agent(model, all_tools)

        # --- Interaja com o agente ---
        print("\nInvocando agente para consulta matemática...")
        math_inputs = {"messages": [("human", "qual é (3 + 5) * 12?")]}
        math_response = await agent.ainvoke(math_inputs)
        print("Resposta Matemática:", math_response['messages'][-1].content)

        print("\nInvocando agente para consulta do clima...")
        weather_inputs = {"messages": [("human", "qual é o clima em nova york?")]}
        weather_response = await agent.ainvoke(weather_inputs)
        print("Resposta do Clima:", weather_response['messages'][-1].content)

        # --- Exemplo: Obtendo um prompt ---
        # print("\nObtendo prompt do servidor de matemática...")
        # prompt_messages = await client.get_prompt(
        #     server_name="math_service", # Use o nome definido nas conexões
        #     prompt_name="configure_assistant",
        #     arguments={"skills": "aritmética básica"}
        # )
        # print("Prompt:", prompt_messages)


if __name__ == "__main__":
    # Inicie primeiro o servidor de clima em um terminal separado:
    # python weather_server.py
    # Em seguida, execute este script do cliente:
    asyncio.run(main())

Salve isso como multi_client_app.py.

Para Executar:

Inicie o servidor de clima em um terminal: python weather_server.py
Execute o aplicativo multi-cliente em outro terminal: python multi_client_app.py

O MultiServerMCPClient iniciará o subprocesso math_server.py (stdio) e se conectará ao weather_server.py em execução (sse). Ele agrega ferramentas (add, multiply, get_weather) que então ficam disponíveis para o agente LangGraph.

Integração com o Servidor de API LangGraph

Você pode implantar um agente LangGraph usando ferramentas MCP como um serviço de API persistente usando langgraph deploy. A chave é gerenciar corretamente o ciclo de vida do MultiServerMCPClient dentro do contexto da aplicação LangGraph.

Crie um arquivo graph.py:

# graph.py
from contextlib import asynccontextmanager
import os
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI # Ou Anthropic, etc.

# --- IMPORTANTE: Atualize os caminhos ---
# Supondo que os servidores estejam relativos a onde o servidor LangGraph é executado
math_server_script_path = os.path.abspath("math_server.py")
# ---

# Defina conexões (certifique-se de que os caminhos/URLs estejam corretos para o ambiente do servidor)
server_connections = {
    "math_service": {
        "transport": "stdio",
        "command": "python",
        "args": [math_server_script_path],
    },
    "weather_service": {
        "transport": "sse",
        "url": "http://localhost:8000/sse", # O servidor de clima deve estar rodando independentemente
    }
}

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

# Use um gerenciador de contexto assíncrono para gerenciar a configuração/desmontagem do cliente
@asynccontextmanager
async def lifespan(_app): # LangGraph espera essa estrutura para gestão de ciclo de vida
    async with MultiServerMCPClient(server_connections) as client:
        print("Cliente MCP inicializado dentro do ciclo de vida.")
        # Crie o agente *dentro* do contexto onde o cliente está ativo
        agent = create_react_agent(model, client.get_tools())
        yield {"agent": agent} # Torne o agente disponível

# Não é necessário uma definição separada do gráfico principal se o ciclo de vida retornar isso

Configure seu langgraph.json (ou pyproject.toml sob [tool.langgraph]) para usar esta definição de gráfico com o gerenciador de ciclo de vida:

// langgraph.json (exemplo)
{
  "dependencies": ["."], // Ou especifique pacotes necessários
  "graphs": {
    "my_mcp_agent": {
      "entrypoint": "graph:agent", // Refere-se à chave retornada pelo ciclo de vida
      "lifespan": "graph:lifespan"
    }
  }
}

Agora, quando você executar langgraph up, a função lifespan será executada, iniciando o MultiServerMCPClient (e o servidor matemático stdio). O agente criado dentro desse contexto será servido pelo LangGraph. Lembre-se de que o servidor de clima SSE ainda precisa ser executado separadamente.

Transportes de Servidor (stdio vs. SSE)

stdio:

Comunicação: Via os streams de entrada e saída padrão do processo do servidor.
Prós: Configuração simples para desenvolvimento local; o cliente gerencia o ciclo de vida do servidor. Sem rede envolvida.
Contras: Rigorosamente acoplado; menos adequado para sistemas distribuídos ou servidores não-Python. Exige configuração de command e args.

sse (Eventos Enviados pelo Servidor):

Comunicação: Sobre HTTP usando o protocolo SSE. O servidor roda como um serviço web (geralmente utilizando FastAPI/Uvicorn implicitamente).
Prós: Protocolo web padrão; adequado para servidores em rede/remotos, potencialmente implementados em diferentes linguagens. O servidor roda independentemente.
Contras: Exige que o servidor esteja rodando separadamente. Necessita configuração de url.

Escolha o transporte com base nas suas necessidades de implantação.

Configuração Avançada do Cliente para a Configuração do langchain-mcp-adapters

Os dicionários StdioConnection e SSEConnection dentro do MultiServerMCPClient aceitam parâmetros adicionais opcionais para um controle mais fino:

Stdio: env (variáveis de ambiente personalizadas para o subprocesso), cwd (diretório de trabalho), encoding, encoding_error_handler, session_kwargs (passados para mcp.ClientSession).
SSE: headers (cabeçalhos HTTP personalizados), timeout (timeout de conexão HTTP), sse_read_timeout, session_kwargs.

Consulte a definição de MultiServerMCPClient em langchain_mcp_adapters/client.py para detalhes.

Conclusão (100 palavras)

A biblioteca langchain-mcp-adapters efetivamente fecha a lacuna entre o Protocolo de Contexto do Modelo padronizado e o flexível ecossistema LangChain. Ao fornecer o MultiServerMCPClient e conversão automática de ferramentas, permite que desenvolvedores incorporem facilmente ferramentas diversas e compatíveis com o MCP em seus agentes LangChain e aplicações LangGraph.

O fluxo de trabalho central envolve:

Definição de ferramentas (e opcionalmente prompts) em um servidor MCP usando @mcp.tool().
Configuração do MultiServerMCPClient com detalhes de conexão (stdio ou sse) para cada servidor.
Uso do gerenciador de contexto do cliente (async with ...) para conectar e buscar ferramentas via client.get_tools().
Passagem das ferramentas compatíveis com LangChain recuperadas para seu agente (create_react_agent ou agentes personalizados).

Isso permite construir aplicações de IA poderosas e modulares que aproveitam ferramentas externas especializadas através de um protocolo padronizado. Explore os exemplos e testes dentro do repositório para mais insights.