Agente de IA con RAG y MCP: integración de herramientas externas

Construyendo un agente de IA con RAG y herramientas externas vía MCP

Los agentes de IA modernos van más allá de la generación de texto: interactúan con APIs externas, gestionan el estado y toman decisiones basadas en el contexto. En este artículo, construiremos un agente que combina RAG para trabajar con documentación interna y el Protocolo de Contexto de Modelo (MCP) para acceder a servicios externos: el tiempo y los issues de GitHub. La arquitectura se basa en LangGraph y FastMCP, lo que permite escalar el sistema sin escribir adaptadores personalizados.

Componentes arquitectónicos: ReAct, RAG y MCP

El agente de IA en nuestro caso opera en el ciclo ReAct: el modelo analiza la consulta → selecciona una herramienta → obtiene el resultado → repite el ciclo hasta la respuesta final. Esto difiere de las cadenas lineales, donde la secuencia de pasos está rígidamente predefinida. LangGraph se usa para soportar la toma de decisiones: proporciona gestión explícita del estado y gráficos de transición.

Aquí, RAG no actúa como un módulo separado, sino como una de las herramientas del agente. Cuando un usuario pregunta sobre documentación interna, el agente llama a search_documentation en lugar de depender de la memoria paramétrica del modelo. Esto reduce el riesgo de alucinaciones y aumenta la precisión de las respuestas.

El Protocolo de Contexto de Modelo (MCP) estandariza la conexión de herramientas externas. En lugar de escribir manualmente envoltorios para cada API, ejecutamos un servidor MCP que proporciona herramientas vía JSON-RPC. El agente se conecta al servidor mediante un cliente, y todas las herramientas quedan disponibles automáticamente. Esto es especialmente útil en producción, donde el número de integraciones crece rápidamente.

Pila tecnológica y justificación de la elección

La elección de tecnologías está dictada por requisitos de escalabilidad, soporte asíncrono y compatibilidad con MCP:

• Python 3.11+: soporte para async/await moderno, ecosistema estable.
• LangGraph + LangChain: LangGraph proporciona control sobre el estado y el ciclo ReAct, LangChain: abstracciones para modelos y herramientas.
• FastMCP: biblioteca de Python para crear servidores MCP rápidamente mediante decoradores. La versión 2.0 se desarrolla activamente e incluye capacidades de cliente.
• ChromaDB: DB vectorial ligera y embebible, ideal para prototipos.
• OpenAI text-embedding-3-small: relación óptima precio/rendimiento para embeddings.
• Claude o GPT-4: cualquier modelo con soporte para function calling.

Implementando el servidor MCP para herramientas externas

El primer paso es crear un servidor MCP que proporciona dos herramientas: obtener el tiempo y crear un issue de GitHub. El servidor se construye sobre FastMCP y se ejecuta como un subproceso con transporte stdio.

# tools_server.py
import os
import httpx
from fastmcp import FastMCP

mcp = FastMCP("ExternalTools")

@mcp.tool()
async def get_weather(city: str) -> str:
    async with httpx.AsyncClient() as client:
        geo_response = await client.get(
            "https://geocoding-api.open-meteo.com/v1/search",
            params={"name": city, "count": 1, "language": "en", "format": "json"},
            timeout=10.0
        )
        geo_data = geo_response.json()
        if not geo_data.get("results"):
            return f"Gorod '{city}' not found"
        
        location = geo_data["results"][0]
        lat, lon = location["latitude"], location["longitude"]
        
        weather_response = await client.get(
            "https://api.open-meteo.com/v1/forecast",
            params={"latitude": lat, "longitude": lon, "current_weather": True},
            timeout=10.0
        )
        current = weather_response.json()["current_weather"]
        return f"Pogoda in {location['name']}: {current['temperature']}°C, veter {current['windspeed']} km/ch"

@mcp.tool()
async def create_github_issue(repo: str, title: str, body: str, labels: list[str] | None = None) -> str:
    github_token = os.environ.get("GITHUB_TOKEN")
    if not github_token:
        return "Error: GITHUB_TOKEN not ustanovlen"
    
    url = f"https://api.github.com/repos/{repo}/issues"
    headers = {"Authorization": f"Bearer {github_token}", "Accept": "application/vnd.github+json"}
    payload = {"title": title, "body": body}
    if labels: payload["labels"] = labels
    
    async with httpx.AsyncClient() as client:
        response = await client.post(url, json=payload, headers=headers)
        if response.status_code == 201:
            return f"Issue sozdan: {response.json()['html_url']}"
        else:
            return f"Error: {response.status_code}"

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

Características clave de la implementación:

• El decorador @mcp.tool() genera automáticamente el esquema JSON para los parámetros basado en anotaciones de tipo y docstring.
• Se usa un cliente HTTP asíncrono para solicitudes no bloqueantes.
• El manejo de errores está integrado en la lógica de cada herramienta: el agente recibe mensajes legibles por humanos en lugar de excepciones.
• El transporte stdio permite al cliente ejecutar el servidor como proceso hijo y comunicarse vía stdin/stdout.

Configurando el módulo RAG como herramienta del agente

El sistema RAG indexa documentos PDF y proporciona una función de búsqueda mediante una herramienta compatible con LangChain. Importante: esta herramienta se registra en el mismo conjunto que las herramientas MCP, permitiendo al agente seleccionarla junto con las APIs externas.

# rag_tool.py
import os
from pathlib import Path
from langchain_openai import OpenAIEmbeddings
from langchain_community.document_loaders import PyPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain.tools import tool

CHROMA_PATH = "./chroma_db"
DATA_PATH = "./data"

@tool
def search_documentation(query: str) -> str:
    """Ischet relevantnye fragmenty in documentation by request."""
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
    vectorstore = Chroma(persist_directory=CHROMA_PATH, embedding_function=embeddings)
    docs = vectorstore.similarity_search(query, k=3)
    return "\n\n".join([f"Withtranitsa {d.metadata.get('page', 'N/A')}: {d.page_content[:500]}..." for d in docs])

# Function initsializatsii database (vyzyvaetsya when starte agenta)
def initialize_vector_store(force_reload=False):
    if os.path.exists(CHROMA_PATH) and not force_reload:
        return
    
    documents = []
    for pdf_file in Path(DATA_PATH).glob("*.pdf"):
        loader = PyPDFLoader(str(pdf_file))
        documents.extend(loader.load())
    
    if not documents:
        return
    
    splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
    chunks = splitter.split_documents(documents)
    
    Chroma.from_documents(chunks, OpenAIEmbeddings(model="text-embedding-3-small"), persist_directory=CHROMA_PATH)

Integrando el cliente MCP y ensamblando el agente en LangGraph

La etapa final es ensamblar el agente que combina herramientas MCP y RAG. Usamos langchain-mcp-adapters para convertir herramientas MCP al formato LangChain, luego las registramos junto con la herramienta RAG en el grafo.

# agent.py
from langgraph.graph import StateGraph, END
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage, AIMessage
from langchain_mcp_adapters import McpToolNode
from rag_tool import search_documentation
import subprocess
import asyncio

# Run MCP-server how podprotsessa
mcp_process = subprocess.Popen(
    ["python", "tools_server.py"],
    stdin=subprocess.PIPE,
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE
)

# Withbudynek MCP-client and adaptera narzędzieov
mcp_node = McpToolNode(
    server_command=["python", "tools_server.py"],
    transport="stdio"
)

# Receiving list narzędzieov from MCP-server
mcp_tools = asyncio.run(mcp_node.get_tools())
all_tools = [*mcp_tools, search_documentation]

# Initialization modeli and graph
model = ChatAnthropic(model="claude-3-5-sonnet-20240620").bind_tools(all_tools)

# Withstoyanie graph
class AgentState(TypedDict):
    messages: Annotated[list, operator.add]

# Uzly graph
def call_model(state):
    response = model.invoke(state["messages"])
    return {"messages": [response]}

def should_continue(state):
    last_message = state["messages"][-1]
    return "continue" if last_message.tool_calls else "end"

# Postroenie graph
workflow = StateGraph(AgentState)
workflow.add_node("agent", call_model)
workflow.add_node("tools", mcp_node)
workflow.set_entry_point("agent")
workflow.add_conditional_edges("agent", should_continue, {"continue": "tools", "end": END})
workflow.add_edge("tools", "agent")

app = workflow.compile()

# Example run
result = app.invoke({"messages": [HumanMessage(content="Uznay pogodu in Londone and sozday issue in repo/my-project with etim rezultatom")]})
print(result["messages"][-1].content)

Lo importante:

• MCP elimina la necesidad de escribir adaptadores personalizados para cada nueva herramienta: solo agrega un decorador @mcp.tool() en el servidor.
• RAG se integra como una herramienta regular, permitiendo al agente elegir entre buscar en la documentación y llamar a una API externa.
• LangGraph proporciona gestión transparente del estado y control del bucle ReAct, crucial para tareas complejas de múltiples pasos.
• La implementación asíncrona de herramientas evita bloquear el hilo principal del agente.
• El transporte stdio simplifica el despliegue: el servidor y el agente pueden ejecutarse en el mismo contenedor o en la misma máquina sin dependencias de red.

— Editorial Team