Agent IA avec RAG et MCP : intégration d'outils externes

Construire un agent IA avec RAG et outils externes via MCP

Les agents IA modernes vont au-delà de la génération de texte — ils interagissent avec des API externes, gèrent l'état et prennent des décisions en fonction du contexte. Dans cet article, nous allons construire un agent qui combine RAG pour travailler avec la documentation interne et le Model Context Protocol (MCP) pour accéder à des services externes : météo et GitHub Issues. L'architecture est basée sur LangGraph et FastMCP, permettant au système de scaler sans écrire d'adaptateurs personnalisés.

Composants architecturaux : ReAct, RAG et MCP

L'agent IA dans notre cas fonctionne selon le cycle ReAct : le modèle analyse la requête → sélectionne un outil → obtient le résultat → répète le cycle jusqu'à la réponse finale. Cela diffère des chaînes linéaires, où la séquence d'étapes est rigoureusement prédéfinie. LangGraph est utilisé pour supporter la prise de décision — il fournit une gestion explicite de l'état et des graphes de transition.

RAG intervient ici non pas comme un module séparé, mais comme l'un des outils de l'agent. Quand un utilisateur pose une question sur la documentation interne, l'agent appelle search_documentation au lieu de s'appuyer sur la mémoire paramétrique du modèle. Cela réduit le risque d'hallucinations et améliore la précision des réponses.

Le Model Context Protocol (MCP) standardise la connexion d'outils externes. Au lieu d'écrire manuellement des wrappers pour chaque API, nous lançons un serveur MCP qui fournit les outils via JSON-RPC. L'agent se connecte au serveur via un client, et tous les outils deviennent disponibles automatiquement. C'est particulièrement utile en production, où le nombre d'intégrations croît rapidement.

Stack technique et justifications des choix

Le choix des technologies est dicté par les exigences de scalabilité, de support asynchrone et de compatibilité MCP :

• Python 3.11+ : prise en charge de async/await moderne, écosystème stable.
• LangGraph + LangChain : LangGraph offre un contrôle sur l'état et le cycle ReAct, LangChain — des abstractions pour les modèles et les outils.
• FastMCP : bibliothèque Python pour créer rapidement des serveurs MCP via des décorateurs. La version 2.0 est activement développée et inclut des capacités client.
• ChromaDB : base de données vectorielle légère et embarquée, idéale pour les prototypes.
• OpenAI text-embedding-3-small : ratio prix/performance optimal pour les embeddings.
• Claude ou GPT-4 : n'importe quel modèle avec support de function calling.

Implémentation du serveur MCP pour les outils externes

La première étape consiste à créer un serveur MCP qui fournit deux outils : obtenir la météo et créer un GitHub Issue. Le serveur est basé sur FastMCP et s'exécute en tant que sous-processus avec transport 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")

Caractéristiques clés de l'implémentation :

• Le décorateur @mcp.tool() génère automatiquement le schéma JSON des paramètres en se basant sur les annotations de type et la docstring.
• Un client HTTP asynchrone est utilisé pour des requêtes non bloquantes.
• La gestion d'erreurs est intégrée dans la logique de chaque outil — l'agent reçoit des messages lisibles par l'humain au lieu d'exceptions.
• Le transport stdio permet au client d'exécuter le serveur en tant que processus enfant et de communiquer via stdin/stdout.

Mise en place du module RAG en tant qu'outil d'agent

Le système RAG indexe des documents PDF et fournit une fonction de recherche via un outil compatible LangChain. Importamment, cet outil est enregistré dans le même pool que les outils MCP, permettant à l'agent de le sélectionner aux côtés des API externes.

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

Intégration du client MCP et assemblage de l'agent dans LangGraph

L'étape finale consiste à assembler l'agent qui combine les outils MCP et RAG. Nous utilisons langchain-mcp-adapters pour convertir les outils MCP au format LangChain, puis les enregistrons avec l'outil RAG dans le graphe.

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

Points importants :

• MCP élimine le besoin d'écrire des adaptateurs personnalisés pour chaque nouvel outil — il suffit d'ajouter un décorateur @mcp.tool() sur le serveur.
• RAG s'intègre comme un outil ordinaire, permettant à l'agent de choisir entre rechercher dans la documentation et appeler une API externe.
• LangGraph fournit une gestion transparente de l'état et un contrôle de la boucle ReAct, ce qui est crucial pour les tâches complexes en plusieurs étapes.
• L'implémentation asynchrone des outils évite de bloquer le thread principal de l'agent.
• Le transport stdio simplifie le déploiement — le serveur et l'agent peuvent s'exécuter dans le même conteneur ou sur la même machine sans dépendances réseau.

— Editorial Team