Einen KI-Agenten mit RAG und externen Tools über MCP aufbauen
Moderne KI-Agenten gehen über reine Textgenerierung hinaus – sie interagieren mit externen APIs, verwalten Zustände und treffen Entscheidungen basierend auf dem Kontext. In diesem Artikel bauen wir einen Agenten auf, der RAG für die Arbeit mit interner Dokumentation und das Model Context Protocol (MCP) für den Zugriff auf externe Dienste kombiniert: Wetter und GitHub Issues. Die Architektur basiert auf LangGraph und FastMCP, was das System skalierbar macht, ohne eigene Adapter schreiben zu müssen.
Architekturkomponenten: ReAct, RAG und MCP
Der KI-Agent arbeitet in unserem Fall nach dem ReAct-Zyklus: Das Modell analysiert die Anfrage → wählt ein Tool aus → erhält das Ergebnis → wiederholt den Zyklus bis zur finalen Antwort. Das unterscheidet sich von linearen Chains, bei denen die Schrittfolge starr vorgegeben ist. LangGraph unterstützt die Entscheidungsfindung – es bietet explizite Zustandsverwaltung und Übergangsgraphen.
RAG fungiert hier nicht als separates Modul, sondern als eines der Tools des Agenten. Wenn ein Nutzer nach interner Dokumentation fragt, ruft der Agent search_documentation auf, anstatt auf das parametrische Gedächtnis des Modells zu setzen. Das reduziert das Risiko von Halluzinationen und erhöht die Genauigkeit der Antworten.
Das Model Context Protocol (MCP) standardisiert die Anbindung externer Tools. Statt manuell Wrapper für jede API zu schreiben, betreiben wir einen MCP-Server, der Tools über JSON-RPC bereitstellt. Der Agent verbindet sich über einen Client mit dem Server, und alle Tools stehen automatisch zur Verfügung. Das ist besonders nützlich in der Produktion, wo die Anzahl der Integrationen schnell wächst.
Tech-Stack und Auswahlbegründung
Die Technologieauswahl richtet sich nach Anforderungen an Skalierbarkeit, asynchrone Unterstützung und MCP-Kompatibilität:
- Python 3.11+: Unterstützung für modernes async/await, stabiles Ökosystem.
- LangGraph + LangChain: LangGraph bietet Kontrolle über Zustand und ReAct-Zyklus, LangChain – Abstraktionen für Modelle und Tools.
- FastMCP: Python-Bibliothek zum schnellen Erstellen von MCP-Servern über Dekoratoren. Version 2.0 wird aktiv entwickelt und umfasst Client-Fähigkeiten.
- ChromaDB: Leichtgewichtige eingebettete Vektordatenbank, ideal für Prototypen.
- OpenAI text-embedding-3-small: Optimales Preis-Leistungs-Verhältnis für Embeddings.
- Claude oder GPT-4: Jedes Modell mit Function-Calling-Unterstützung.
Implementierung des MCP-Servers für externe Tools
Der erste Schritt ist die Erstellung eines MCP-Servers, der zwei Tools bereitstellt: Wetter abrufen und GitHub Issue erstellen. Der Server basiert auf FastMCP und läuft als Subprozess mit stdio-Transport.
# 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")Wichtige Merkmale der Implementierung:
- Der
@mcp.tool()-Dekorator erzeugt automatisch das JSON-Schema für Parameter basierend auf Typannotationen und Docstring. - Ein asynchroner HTTP-Client wird für nicht-blockierende Anfragen verwendet.
- Fehlerbehandlung ist in die Logik jedes Tools integriert – der Agent erhält lesbare Nachrichten statt Exceptions.
- Stdio-Transport erlaubt es dem Client, den Server als Child-Prozess laufen zu lassen und über stdin/stdout zu kommunizieren.
Einrichtung des RAG-Moduls als Agent-Tool
Das RAG-System indexiert PDF-Dokumente und stellt eine Suchfunktion über ein LangChain-kompatibles Tool bereit. Wichtig: Dieses Tool wird im selben Pool wie die MCP-Tools registriert, sodass der Agent es neben externen APIs wählen kann.
# 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)Integration des MCP-Clients und Zusammenbau des Agenten in LangGraph
Die abschließende Phase ist der Zusammenbau des Agenten, der MCP-Tools und RAG kombiniert. Wir nutzen langchain-mcp-adapters, um MCP-Tools in LangChain-Format zu konvertieren, und registrieren sie zusammen mit dem RAG-Tool im Graphen.
# 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)Wichtig:
- MCP macht das Schreiben benutzerdefinierter Adapter für jedes neue Tool überflüssig – einfach einen
@mcp.tool()-Dekorator auf dem Server hinzufügen. - RAG integriert sich als reguläres Tool, sodass der Agent zwischen der Suche in der Dokumentation und dem Aufruf einer externen API wählen kann.
- LangGraph sorgt für transparente Zustandsverwaltung und ReAct-Schleifensteuerung, was für komplexe Mehrschritt-Aufgaben entscheidend ist.
- Asynchrone Tool-Implementierung verhindert das Blockieren des Haupt-Threads des Agenten.
- Stdio-Transport vereinfacht die Bereitstellung – Server und Agent können im selben Container oder auf derselben Maschine laufen, ohne Netzwerkabhängigkeiten.
— Editorial Team
Noch keine Kommentare.