AI agent s RAG a MCP: integrace externích nástrojů

# Sestavení AI agenta s RAG a externími nástroji prostřednictvím MCP

Moderní AI agenti překračují hranice generování textu – interagují s externími API, spravují stav a rozhodují se na základě kontextu. V tomto článku sestavíme agenta, který kombinuje RAG pro práci s interní dokumentací a Model Context Protocol (MCP) pro přístup k externím službám: počasí a GitHub Issues. Architektura je postavena na LangGraph a FastMCP, což umožňuje škálovat systém bez psaní vlastních adaptérů.

Architekturické komponenty: ReAct, RAG a MCP

AI agent v našem případě pracuje v cyklu ReAct: Model analyzuje požadavek → vybere nástroj → získá výsledek → opakuje cyklus až do finální odpovědi. To se liší od lineárních řetězců (chains), kde je posloupnost kroků pevně daná. Pro podporu rozhodování se používá LangGraph – zajišťuje explicitní správu stavu a grafu přechodů.

RAG zde funguje ne jako samostatný modul, ale jako jeden z nástrojů agenta. Když uživatel ptá na interní dokumentaci, agent zavolá search_documentation, místo aby se spoléhal na parametrickou paměť modelu. To snižuje riziko halucinací a zvyšuje přesnost odpovědí.

Model Context Protocol (MCP) standardizuje připojení externích nástrojů. Místo ručního psaní obalů pro každé API spustíme MCP server, který poskytuje nástroje přes JSON-RPC. Agent se připojí k serveru přes klienta a všechny nástroje se stávají automaticky dostupné. To je obzvláště užitečné v produkci, kde počet integrací rychle roste.

Technologický stack a odůvodnění volby

Volba technologií vychází z požadavků na škálovatelnost, podporu asynchronnosti a kompatibilitu s MCP:

• Python 3.11+: podpora moderního async/await, stabilní ekosystém.
• LangGraph + LangChain: LangGraph dává kontrolu nad stavem a cyklem ReAct, LangChain – abstrakce pro modely a nástroje.
• FastMCP: Python knihovna pro rychlé vytváření MCP serverů přes dekorátory. Verze 2.0 se aktivně vyvíjí a zahrnuje klientské funkce.
• ChromaDB: lehká vestavěná vektorová DB, ideální pro prototypy.
• OpenAI text-embedding-3-small: optimální poměr cena/výkon pro embeddingy.
• Claude nebo GPT-4: jakýkoli model s podporou function calling.

Implementace MCP serveru pro externí nástroje

První krok – vytvoření MCP serveru, který poskytuje dva nástroje: získání počasí a vytvoření GitHub Issue. Server je postaven na FastMCP a spouští se jako podproces s transportem 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")

Klíčové vlastnosti implementace:

• Dekorátor @mcp.tool() automaticky generuje JSON schému parametrů na základě anotací typů a docstringu.
• Používá se asynchronní HTTP klient pro neblokující požadavky.
• Zpracování chyb je vestavěné do logiky každého nástroje – agent dostává čitelná hlášení místo výjimek.
• Transport stdio umožňuje klientovi spustit server jako podproces a komunikovat přes stdin/stdout.

Nastavení RAG modulu jako nástroje agenta

RAG systém indexuje PDF dokumenty a poskytuje funkci vyhledávání přes nástroj kompatibilní s LangChain. Důležité je, že tento nástroj se registruje ve stejném poolu jako MCP nástroje, což umožňuje agentovi ho vybrat na rovině s externími API.

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

Integrace MCP klienta a sestavení agenta v LangGraph

Finální etapa – sestavení agenta, který spojuje MCP nástroje a RAG. Používáme langchain-mcp-adapters pro převod MCP nástrojů do formátu LangChain, poté je registrováme společně s RAG nástrojem do grafu.

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

Co je důležité:

• MCP klient spouští server jako podproces a spravuje jeho životní cyklus.
• Všechny nástroje (MCP + RAG) se spojí do jediného seznamu a předají modelu přes .bind_tools().
• Graf obsahuje dva uzly: agent (generování myšlenky/akce) a tools (vykonání nástroje).
• Přechod mezi uzly řídí podmínka: pokud jsou tool_calls – jdeme do tools, jinak končíme.

Co je důležité

• MCP eliminuje potřebu psát vlastní adaptéry pro každý nový nástroj – stačí přidat dekorátor @tool() na serveru.
• RAG se integruje jako běžný nástroj, což umožňuje agentovi volit mezi hledáním v dokumentaci a voláním externího API.
• LangGraph zajišťuje průhlednou správu stavu a cyklu ReAct, což je klíčové pro složité vícekrokové úkoly.
• Asynchronní implementace nástrojů zabraňuje blokování hlavního vlákna agenta.
• Transport stdio zjednodušuje nasazení – server a agent mohou běžet v jednom kontejneru nebo na jedné mašině bez síťových závislostí.

— Editorial Team