AI-agent z RAG i MCP: integracja zewnętrznych narzędzi

# Budowa agenta AI z RAG i zewnętrznymi narzędziami za pomocą MCP

Nowoczesne agenty AI wykraczają poza generowanie tekstu — interagują z zewnętrznymi API, zarządzają stanem i podejmują decyzje na podstawie kontekstu. W tym artykule zbudujemy agenta, który łączy RAG do pracy z wewnętrzną dokumentacją i Model Context Protocol (MCP) do dostępu do zewnętrznych usług: pogody i GitHub Issues. Architektura opiera się na LangGraph i FastMCP, co pozwala skalować system bez pisania niestandardowych adapterów.

Komponenty architektoniczne: ReAct, RAG i MCP

AI-agent w naszym przypadku działa w pętli ReAct: Model analizuje zapytanie → wybiera narzędzie → pobiera wynik → powtarza pętlę aż do ostatecznej odpowiedzi. To różni się od liniowych łańcuchów (chains), gdzie sekwencja kroków jest sztywno określona. Do obsługi podejmowania decyzji używamy LangGraph — zapewnia on jawne zarządzanie stanem i grafem przejść.

RAG tutaj działa nie jako oddzielny moduł, lecz jako jedno z narzędzi agenta. Gdy użytkownik pyta o wewnętrzną dokumentację, agent wywołuje search_documentation, zamiast polegać na parametrycznej pamięci modelu. To zmniejsza ryzyko halucynacji i zwiększa dokładność odpowiedzi.

Model Context Protocol (MCP) standaryzuje podłączanie zewnętrznych narzędzi. Zamiast ręcznego pisania wrapperów pod każde API, uruchamiamy MCP-serwer, który udostępnia narzędzia przez JSON-RPC. Agent podłącza się do serwera przez klienta, a wszystkie narzędzia stają się dostępne automatycznie. To szczególnie przydatne w produkcji, gdzie liczba integracji szybko rośnie.

Stack technologiczny i uzasadnienie wyboru

Wybór technologii wynika z wymagań skalowalności, obsługi asynchroniczności i kompatybilności z MCP:

• Python 3.11+: obsługa nowoczesnego async/await, stabilne ekosystem.
• LangGraph + LangChain: LangGraph daje kontrolę nad stanem i pętlą ReAct, LangChain — abstrakcje dla modeli i narzędzi.
• FastMCP: biblioteka Python do szybkiego tworzenia serwerów MCP za pomocą dekoratorów. Wersja 2.0 jest aktywnie rozwijana i zawiera możliwości klienckie.
• ChromaDB: lekka, wbudowana wektorowa baza danych, idealna do prototypów.
• OpenAI text-embedding-3-small: optymalne stosunek cena/jakość embeddingów.
• Claude lub GPT-4: dowolny model z obsługą function calling.

Implementacja serwera MCP dla zewnętrznych narzędzi

Pierwszy krok — stworzenie serwera MCP, który udostępnia dwa narzędzia: pobieranie pogody i tworzenie GitHub Issue. Serwer jest budowany na FastMCP i uruchamiany jako podproces z 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")

Kluczowe cechy implementacji:

• Dekorator @mcp.tool() automatycznie generuje schemat JSON parametrów na podstawie adnotacji typów i docstring.
• Używa asynchronicznego klienta HTTP do nieblokujących zapytań.
• Obsługa błędów jest wbudowana w logikę każdego narzędzia — agent otrzymuje czytelne dla człowieka komunikaty zamiast wyjątków.
• Transport stdio pozwala klientowi uruchamiać serwer jako proces potomny i komunikować się przez stdin/stdout.

Konfiguracja modułu RAG jako narzędzia agenta

System RAG indeksuje dokumenty PDF i udostępnia funkcję wyszukiwania przez narzędzie kompatybilne z LangChain. Ważne, że to narzędzie jest rejestrowane w tym samym puli co narzędzia MCP, co pozwala agentowi wybierać je na równi z zewnętrznymi 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)

Integracja klienta MCP i budowa agenta w LangGraph

Ostatni etap — budowa agenta, który łączy narzędzia MCP i RAG. Używamy langchain-mcp-adapters do konwersji narzędzi MCP na format LangChain, a następnie rejestrujemy je razem z narzędziem RAG w grafie.

# 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 ważne:

• Klient MCP uruchamia serwer jako podproces i zarządza jego cyklem życia.
• Wszystkie narzędzia (MCP + RAG) są łączone w jedną listę i przekazywane modelowi przez .bind_tools().
• Graf zawiera dwa węzły: agent (generowanie myśli/działania) i tools (wykonywanie narzędzia).
• Przejście między węzłami jest sterowane warunkiem: jeśli są tool_calls — idziemy do tools, w przeciwnym razie kończymy.

Co ważne

• MCP eliminuje potrzebę pisania niestandardowych adapterów dla każdego nowego narzędzia — wystarczy dodać dekorator @tool() na serwerze.
• RAG integruje się jak zwykłe narzędzie, co pozwala agentowi wybierać między wyszukiwaniem w dokumentacji a wywołaniem zewnętrznego API.
• LangGraph zapewnia przejrzyste zarządzanie stanem i pętlą ReAct, co jest kluczowe dla złożonych zadań wieloetapowych.
• Asynchroniczna implementacja narzędzi zapobiega blokowaniu głównego wątku agenta.
• Transport stdio upraszcza wdrożenie — serwer i agent mogą działać w jednym kontenerze lub na jednej maszynie bez zależności sieciowych.

— Editorial Team