Middleware
Middleware führt vor und nach der Agent-Ausführung eigene Logik aus: Dokumente anhängen, Kontext anreichern, Eingaben validieren und Antworten transformieren.
Auf dieser Seite
Was ist Middleware?
Hooks, die vor und nach der Verarbeitung einer Anfrage durch den Agenten ausgeführt werden
Middleware sind Python-Funktionen, die Anfragen abfangen, bevor sie den Agenten erreichen, und Antworten, bevor sie zurückgegeben werden. Damit lassen sich Dokumente anhängen, Kontext anreichern, Eingaben validieren, Interaktionen protokollieren oder Antworten transformieren.
Erstelle im Verzeichnis middleware/ eine Datei mit demselben Namen wie der Agent. So gilt middleware/assistant.py beispielsweise für den Agenten, dessen YAML name: assistant enthält – selbst wenn diese YAML-Datei in einem Unterordner von agents/ liegt. Eine YAML-Konfiguration ist nicht erforderlich.
Ausführungsablauf
Einfache Middleware
"""Middleware for the assistant agent."""
import json
from datetime import datetime
from typing import Any, Dict
async def before(content: Dict[str, Any], context: Dict[str, Any]) -> Dict[str, Any]:
"""
Called before the agent runs.
Args:
content: Dict with "role" and "parts" keys
context: Shared run context dict (see Context docs)
Returns:
Modified content to pass to the agent
"""
print(f"Before middleware running for run {context.get('run_id')}")
# Add current timestamp so the agent knows today's date
now = datetime.now()
time_context = f"[Current time: {now.strftime('%A, %B %d, %Y')}]"
content["parts"].insert(0, {"text": time_context})
return content
async def after(response: str, context: Dict[str, Any]) -> str:
"""
Called after the agent completes.
Args:
response: The agent's response text
context: Shared run context dict (see Context docs)
Returns:
Modified response (or original)
"""
print(f"After middleware: response length: {len(response)}")
# Wrap response in structured JSON with metadata
return json.dumps({
"response": response,
"run_id": context.get("run_id"),
"duration_ms": context.get("duration_ms"),
})Beide Hooks sind optional. Es kann nur before(), nur after() oder beides definiert werden.before() muss Content zurückgeben; der Rückgabewert von after() wird zur Antwort.
Das Content-Dictionary
Der Hook before() erhält ein content-Dictionary, das die Nachricht des Benutzers darstellt. Es enthält eine role (immer 'user') und eine Liste von parts. Jeder Part ist ein Dictionary mit Text oder Binärdaten.
# The content dict structure
content = {
"role": "user", # Immer 'user': before() erhält die eingehende Nachricht
"parts": [ # List of part dicts
{"text": "Hello, analyze this document"},
{"data": b"...", "mime_type": "application/pdf"},
]
}
# Accessing content parts
for part in content["parts"]:
if "text" in part:
# Text content
text = part["text"]
elif "data" in part:
# Binary file content
mime_type = part["mime_type"]
data = part["data"]Die ursprüngliche Payload
Die rohe Payload der Verbindung ist ebenfalls als context["payload"] verfügbar. Verwende sie für Request-Metadaten, Authentifizierungsfelder, JSON-Bodys von Webhooks, Formularfelder oder GET-Query-Parameter. Verwende content, um den sichtbaren Agent-Input zu ändern.
# The original connector payload is available in context
payload = context.get("payload", {})
# For a JSON webhook body:
# {"auth_token": "abc", "question": "Hello"}
# payload["auth_token"] == "abc"
# payload["question"] == "Hello"
# For a GET webhook:
# <webhook-url>?user_id=123&query=hello
# payload["user_id"] == "123"
# payload["query"] == "hello"Parts erstellen
Erstelle Parts als einfache Dictionaries: mit text für Textinhalte oder mit data und mime_type für Dateien.
# Creating parts as dicts
# Text part
text_part = {"text": "Analyze this document"}
# File from bytes (PDFs, images, audio, video)
with open("document.pdf", "rb") as f:
pdf_part = {"data": f.read(), "mime_type": "application/pdf"}
# Image from bytes
with open("image.png", "rb") as f:
image_part = {"data": f.read(), "mime_type": "image/png"}
# Adding parts to content
content["parts"].append(pdf_part)
content["parts"].insert(0, text_part)Häufige Use Cases
Dokumente dynamisch anhängen
Füge dem Kontext des Agenten abhängig von der Anfrage PDFs, Bilder oder andere Dateien hinzu.
"""Attach documents dynamically based on request context."""
from typing import Any, Dict
async def before(content: Dict[str, Any], context: Dict[str, Any]) -> Dict[str, Any]:
"""Attach relevant documents to the agent context."""
# Get the user's text message
user_text = ""
for part in content["parts"]:
if "text" in part:
user_text = part["text"]
break
# Attach a retrieval document based on topic
if "pricing" in user_text.lower():
with open("docs/pricing.pdf", "rb") as f:
content["parts"].append({
"data": f.read(),
"mime_type": "application/pdf"
})
elif "technical" in user_text.lower():
with open("docs/technical-specs.pdf", "rb") as f:
content["parts"].append({
"data": f.read(),
"mime_type": "application/pdf"
})
return contentKontext und Kundendaten ergänzen
Ergänze die Anfrage um benötigte Kontextinformationen oder Daten aus externen APIs.
"""Add system context and customer data to requests."""
from typing import Any, Dict
import httpx
async def before(content: Dict[str, Any], context: Dict[str, Any]) -> Dict[str, Any]:
"""Prepend system context to the user message."""
# Read request data from the original connector payload.
payload = context.get("payload", {})
customer_id = payload.get("customer_id")
# Fetch customer data from your API
customer_info = ""
if customer_id:
print(f"Fetching customer data for {customer_id}")
async with httpx.AsyncClient() as client:
resp = await client.get(f"https://api.example.com/customers/{customer_id}")
if resp.status_code == 200:
data = resp.json()
customer_info = f"""
Customer: {data['name']}
Plan: {data['plan']}
Account Status: {data['status']}
"""
print(f"Loaded customer: {data['name']} ({data['plan']})")
# Prepend context as text part
content["parts"].insert(0, {
"text": f"[CUSTOMER CONTEXT]\n{customer_info}\n[END CONTEXT]"
})
return contentSauberer Abbruch mit StopProcessing
"""Authentication middleware - stops processing if invalid."""
from typing import Any, Dict
from connic import StopProcessing
async def before(content: Dict[str, Any], context: Dict[str, Any]) -> Dict[str, Any]:
"""Validate request before agent runs."""
# Read the original connector payload. For webhooks, this is the JSON body,
# form fields, or GET query parameters.
payload = context.get("payload", {})
auth_token = payload.get("auth_token")
if not auth_token:
raise StopProcessing("Authentication required")
# Store verified request data in context for prompts and tools.
context["user_id"] = verify_auth_token(auth_token)
return contentStopProcessing und Exceptions: StopProcessing beendet die Verarbeitung mit einer regulären Antwort (der Run wird als abgeschlossen markiert). Normale Exceptions markieren den Run als fehlgeschlagen. Da der Run abgeschlossen wird, stellen automatische ausgehende Verbindungen ihn standardmäßig weiterhin zu; die StopProcessing-Antwort ist dabei die Ausgabe. Übergib publish_outbound=False, um die automatische ausgehende Verbindung für diesen kontrollierten Abbruch zu überspringen. Dieses Flag bricht keinen Aufruf einer ausgehenden Agent-Tool- oder Middleware-Verbindung ab. Derselbe Mechanismus funktioniert für Tools in tools/. Siehe Tools schreiben
Analytics und Logging
"""Send analytics to external service after agent completes."""
import httpx
from typing import Any, Dict
async def after(response: str, context: Dict[str, Any]) -> str:
"""Send analytics after agent completes."""
try:
async with httpx.AsyncClient() as client:
await client.post(
"https://analytics.internal/events",
json={
"event": "agent_run_completed",
"run_id": context.get("run_id"),
"agent": context.get("agent_name"),
"duration_ms": context.get("duration_ms"),
"tokens": context.get("token_usage", {}),
}
)
except Exception:
pass # Don't fail the request if analytics fails
return responseAusgaben über print oder logging aus der Python-Standardbibliothek werden für jeden Run erfasst. Das gilt für beide Hooks, before() und after(). Die Ausgaben erscheinen im Projekt auf dem Tab Logs sowie in der Run-Detailansicht mit der Quelle middleware.before oder middleware.after.
Ausgehende Middleware-Verbindungen
Setze den Modus einer ausgehenden Verbindung auf Middleware, vergib einen Aufrufnamen und rufe anschließend send_connector(action_name, payload) aus before() oder after() auf. Nutze diesen Modus, wenn Code statt des KI-Modells entscheiden soll, ob gesendet wird.
"""Send an urgent result through a configured middleware outbound connector."""
from typing import Any, Dict
from connic.tools import send_connector
async def after(response: str, context: Dict[str, Any]) -> str:
payload = context.get("payload", {})
if payload.get("priority") == "urgent":
await send_connector(
"send_to_ops_slack",
{"text": response},
)
return responseDie Payload muss dem dokumentierten Payload-Schema der ausgehenden Verbindung entsprechen. Connic verwaltet Verbindung, Ziel, Zugangsdaten, Formatierung und Retries serverseitig; die Middleware erhält keinen dieser gespeicherten Werte. Übergib idempotency_key=..., wenn der Code denselben Aufruf möglicherweise wiederholt. Siehe Modi ausgehender Verbindungen.
Antwort transformieren
"""Transform response format for specific use cases."""
import json
from typing import Any, Dict
def after(response: str, context: Dict[str, Any]) -> str:
"""Wrap response in a standard API format."""
return json.dumps({
"success": True,
"data": response,
"metadata": {
"run_id": context.get("run_id"),
"processed_at": context.get("timestamp"),
}
})Das context-Dictionary wird geteilt und ist veränderbar. In before() gesetzte Werte stehen in Prompts über die Syntax {var}, in Tools und im Hook after() zur Verfügung. Alle Details stehen in der Kontext-Dokumentation.
Projekt-Struktur
Sync- und Async-Unterstützung
"""Sync middleware also works."""
from typing import Any, Dict
def before(content: Dict[str, Any], context: Dict[str, Any]) -> Dict[str, Any]:
"""Sync functions are automatically handled."""
# Add a simple prefix to the first text part
for part in content["parts"]:
if "text" in part:
part["text"] = "[Processed] " + part["text"]
break
return content
def after(response: str, context: Dict[str, Any]) -> str:
"""Both sync and async are supported."""
return response.strip()Verwende Async-Funktionen für I/O-Operationen (API-Aufrufe, Datenbankabfragen und Dateizugriffe). Für einfache Transformationen sind Sync-Funktionen geeignet.
Wenn eine Middleware-Funktion eine Exception auslöst, schlägt die gesamte Anfrage fehl. Der Run wird als fehlgeschlagen markiert; bei before() wird der Agent gar nicht ausgeführt. Nutze Middleware als Gatekeeper für Validierung und Authentifizierung.
Standardmäßig wird before() vor der Ausführung einmal ausgeführt. Setze für Agenten vom Typ tool und sequential retry_options.rerun_middleware: true, um den Hook bei einer erneuten Ausführung des Vorgangs erneut auszuführen.
Das Wiederholen fehlgeschlagener KI-Modell-Anfragen, der Wechsel zum Fallback-KI-Modell und die Verarbeitung von Tool-Fehlern durch das KI-Modell finden innerhalb des laufenden Runs statt. Sie führen Middleware nie erneut aus; deshalb wird rerun_middleware für type: llm ignoriert.
Ist die Option aktiviert, wird bei jedem weiteren Versuch der Input aus der ursprünglichen Payload neu erstellt und before() erneut aufgerufen. So kann Middleware dynamischen Kontext aktualisieren, etwa indem sie nach teilweise ausgeführten Tool-Aufrufen einen externen Zustand erneut abruft. Das context-Dictionary behält gespeicherte Werte über die Wiederholungsversuche hinweg. context.get("retry_attempt", 0) zeigt, welcher Versuch gerade läuft. Die after()-Middleware läuft einmal nach Abschluss aller Retries, sofern before() nicht StopProcessing auslöst.