Eu Tentei Usar Langchain com Servidores MCP, Aqui Estão os Passos:

O Protocolo de Contexto do Modelo (MCP) é um protocolo de código aberto desenvolvido pela Anthropic que aborda um desafio fundamental nas aplicações de Modelos de Linguagem de Grande Escala (LLM): seu isolamento de fontes e ferramentas de dados externas. Este tutorial abrangente irá guiá-lo na implementação do MCP com o LangChain, fornecendo o conhecimento necessário para criar aplicações sofisticadas que utilizam ambas as tecnologias de forma eficaz.

💡

Ao implementar testes para aplicações baseadas em API, desenvolvedores e testadores estão cada vez mais recorrendo a ferramentas especializadas como Apidog, uma alternativa abrangente ao Postman que simplifica o ciclo de desenvolvimento de APIs.

botão

Compreendendo o MCP e Seu Propósito

O Protocolo de Contexto do Modelo visa padronizar como aplicações baseadas em LLM se conectam a diversos sistemas externos. Pense no MCP como o "USB-C para IA" - uma interface universal que permite a troca de dados de forma contínua, segura e escalável entre LLMs/agentes de IA e recursos externos.

O MCP emprega uma arquitetura cliente-servidor:

Anfitriões MCP: aplicações de IA que precisam acessar dados externos
Servidores MCP: provedores de dados ou ferramentas que fornecem informações aos anfitriões

O protocolo facilita uma clara separação de preocupações, permitindo que os desenvolvedores criem conectores modulares e reutilizáveis, enquanto mantêm uma segurança robusta por meio de controles de permissão granular.

Arquitetura Técnica

A arquitetura do MCP consiste em três componentes principais:

Servidor: O servidor MCP expõe ferramentas e fontes de dados por meio de uma API padronizada
Cliente: A aplicação cliente se comunica com o servidor para acessar ferramentas e dados
Adaptador: O LangChain fornece adaptadores que simplificam a integração entre servidores MCP e aplicações LLM

O fluxo de comunicação segue este padrão:

A aplicação LangChain solicita execução de dados/ferramentas
O adaptador MCP transforma a solicitação no formato do protocolo MCP
O servidor processa a solicitação e retorna os resultados
O adaptador transforma a resposta de volta em um formato utilizável pelo LangChain

Pré-requisitos

Antes de começarmos, certifique-se de ter o seguinte:

Python 3.8+ instalado
Chave da API da OpenAI (para usar modelos GPT com LangChain)
Familiaridade básica com conceitos do LangChain
Acesso ao terminal (exemplos mostrados no macOS)

Configurando o Ambiente

Primeiro, vamos criar e configurar nosso ambiente de desenvolvimento:

# Criar um ambiente virtual
python3 -m venv MCP_Demo

# Ativar o ambiente virtual
source MCP_Demo/bin/activate

# Instalar pacotes necessários
pip install langchain-mcp-adapters
pip install langchain-openai

# Defina sua chave da API da OpenAI
export OPENAI_API_KEY=seu_api_key

Criando um Servidor MCP Simples

Começaremos construindo um servidor MCP básico que fornece operações matemáticas. Crie um arquivo chamado math_server.py:

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

mcp = FastMCP("Math")

@mcp.tool()
def add(a: int, b: int) -> int:
    """Adiciona dois números"""
    return a + b

@mcp.tool()
def multiply(a: int, b: int) -> int:
    """Multiplica dois números"""
    return a * b

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

Este servidor expõe duas ferramentas matemáticas: add e multiply. A classe FastMCP simplifica a criação de servidores, lidando automaticamente com os detalhes do protocolo. Cada função decorada com @mcp.tool() se torna disponível para os clientes, com a documentação derivada de docstrings.

Implementando o Cliente LangChain

Em seguida, crie um cliente LangChain para interagir com o servidor MCP. Salve isto como client.py:

# Criar parâmetros do servidor para conexão stdio
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")

# Configurar parâmetros do servidor
server_params = StdioServerParameters(
    command="python",
    # Especifique o caminho para o seu arquivo de servidor
    args=["math_server.py"],
)

async def run_agent():
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # Inicializar a conexão
            await session.initialize()
            
            # Carregar ferramentas MCP no formato do LangChain
            tools = await load_mcp_tools(session)
            
            # Criar e executar o agente
            agent = create_react_agent(model, tools)
            agent_response = await agent.ainvoke({"messages": "qual é (3 + 5) x 12?"})
            return agent_response

# Executar a função assíncrona
if __name__ == "__main__":
    result = asyncio.run(run_agent())
    print(result)

Este cliente estabelece uma conexão com o servidor MCP, carrega as ferramentas disponíveis e cria um agente LangChain que pode usar essas ferramentas para resolver problemas.

Executando o Exemplo

Para executar este exemplo:

Inicie o servidor MCP em uma aba do terminal:

python3 math_server.py

Em outra aba do terminal, execute o cliente:

python3 client.py

O cliente invocará o agente LangChain, que irá:

Interpretar a pergunta "(3 + 5) x 12"
Chamar a ferramenta add com os argumentos 3 e 5
Obter o resultado 8
Chamar a ferramenta multiply com os argumentos 8 e 12
Retornar a resposta final: 96

Implementação Avançada do Servidor MCP

Vamos expandir nossa implementação para criar um servidor MCP mais sofisticado que fornece acesso a bancos de dados. Este exemplo demonstra como construir conectores para 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]]:
    """Executa uma consulta SQL no banco de dados e retorna os resultados"""
    try:
        return db_connector.execute_query(query)
    except Exception as e:
        return {"error": str(e)}

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

Integrando Múltiplos Servidores MCP com LangChain

Para aplicações mais complexas, você pode precisar integrar vários servidores MCP. Veja como criar um cliente que se conecta a múltiplos 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

# Defina nossas configurações de servidor
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):
    """Conectar a um único servidor MCP e carregar suas ferramentas"""
    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():
    # Conectar a todos os servidores
    connections = await asyncio.gather(
        *[connect_to_server(server) for server in servers]
    )
    
    try:
        # Coletar todas as ferramentas de todos os servidores
        all_tools = []
        for connection in connections:
            all_tools.extend(connection["tools"])
        
        # Criar o agente com todas as ferramentas
        model = ChatOpenAI(model="gpt-4o")
        agent = create_react_agent(model, all_tools)
        
        # Executar o agente com uma consulta complexa que pode usar múltiplos servidores
        response = await agent.ainvoke({
            "messages": "Encontre os clientes que gastaram mais do que o valor médio do pedido e calcule seu gasto total."
        })
        
        return response
    
    finally:
        # Limpar todas as conexões
        for connection in connections:
            await connection["cleanup"]()

# Executar o agente de múltiplos servidores
if __name__ == "__main__":
    result = asyncio.run(run_multi_server_agent())
    print(result)

Tratamento de Erros e Estratégias de Retorno

Implementações robustas do MCP devem incluir tratamento de erros. Aqui está uma versão aprimorada do cliente que demonstra o tratamento de erros e estratégias de retorno:

# robust_client.py
async def run_agent_with_fallbacks():
    try:
        # Tentar a conexão primária
        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": "qual é (3 + 5) x 12?"})
                except Exception as e:
                    print(f"Erro ao usar ferramentas MCP: {e}")
                    # Retorno para chamada direta ao modelo sem ferramentas
                    return await model.ainvoke([
                        HumanMessage(content="qual é (3 + 5) x 12?")
                    ])
    except Exception as connection_error:
        print(f"Erro de conexão: {connection_error}")
        # Retorno final
        return {"error": "Não foi possível estabelecer conexão com o servidor MCP"}

Considerações de Segurança

Ao implementar o MCP com o LangChain, considere as seguintes melhores práticas de segurança:

Validação de Entrada: Sempre valide as entradas para as ferramentas MCP para evitar ataques de injeção
Permissões de Ferramenta: Implemente permissões granulares para cada ferramenta
Limitação de Taxa: Aplique limites de taxa para evitar abusos das ferramentas
Autenticação: Implemente autenticação adequada entre clientes e servidores

Aqui está um exemplo de implementação de permissões de ferramenta:

from mcp.server.fastmcp import FastMCP, Permission

mcp = FastMCP("SecureTools")

# Definir níveis de permissão
READ_PERMISSION = Permission("read", "Pode ler dados")
WRITE_PERMISSION = Permission("write", "Pode modificar dados")

@mcp.tool(permissions=[READ_PERMISSION])
def get_data(key: str) -> str:
    """Obter dados por chave (requer permissão de leitura)"""
    # Implementação...
    return f"Dados para {key}"

@mcp.tool(permissions=[WRITE_PERMISSION])
def update_data(key: str, value: str) -> bool:
    """Atualizar dados (requer permissão de escrita)"""
    # Implementação...
    return True

Otimização de Performance

Para implantações de produção, considere essas otimizações de performance:

Pool de Conexões: Reutilize conexões MCP em vez de criar novas para cada solicitação
Processamento em Lote: Agrupe chamadas a múltiplas ferramentas sempre que possível
Processamento Assíncrono: Use asyncio para lidar com múltiplas solicitações simultaneamente

Exemplo de pool de conexões:

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):
        # Pré-criar algumas conexões
        for _ in range(3):  # Começar com 3 conexões
            await self._create_connection()
    
    async def _create_connection(self):
        if self.current_connections >= self.max_connections:
            raise Exception("Máximo de conexões atingido")
        
        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)

Testando Implementações MCP

Testes abrangentes são cruciais para implementações confiáveis do MCP. Aqui está uma abordagem de teste 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

Conclusão

O Protocolo de Contexto do Modelo fornece uma estrutura poderosa para conectar aplicações LangChain a ferramentas e fontes de dados externas. Ao padronizar essas conexões, o MCP permite que os desenvolvedores criem agentes de IA sofisticados que podem interagir de forma contínua com seu ambiente.

A combinação das capacidades de agente do LangChain com a conectividade do MCP cria uma base para construir aplicações verdadeiramente poderosas e cientes do contexto. À medida que o ecossistema MCP continua a crescer, podemos esperar que mais servidores e ferramentas pré-construídos surjam, simplificando ainda mais o processo de desenvolvimento.

Este tutorial abordou os conceitos fundamentais e detalhes de implementação do uso do MCP com LangChain, desde a configuração básica até padrões avançados como pool de conexões e tratamento de erros. Seguindo essas práticas, você pode criar aplicações robustas e prontas para produção que aproveitam o melhor de ambas as tecnologias.

Para uma exploração mais aprofundada, considere investigar o crescente ecossistema de servidores MCP disponíveis no GitHub ou contribuir com seus próprios servidores para a comunidade. O futuro dos agentes de IA depende de sua capacidade de alavancar efetivamente ferramentas e dados externos, e o MCP é um passo significativo para tornar essa visão uma realidade.