Stellen Sie sich folgendes Szenario vor: Ein E-Commerce-Shop zur Hochsaison mit 10.000 gleichzeitigen Kundenanfragen, ein Enterprise-RAG-System, das Millionen von Dokumenten durchsuchen muss, oder ein Indie-Entwickler, der eine KI-gestützte Anwendung ohne Vermögen launchen möchte. In diesem Tutorial zeige ich Ihnen, wie Sie die Claude API Tool Use-Funktion meistern und echte Automatisierungsszenarien umsetzen – mit Preisen ab 0,42 US-Dollar pro Million Token und Latenzzeiten unter 50 Millisekunden.
Was ist Tool Use bei Claude?
Tool Use ermöglicht es Claude, während einer Konversation externe Funktionen aufzurufen. Der Agent kann:
- Datenbankabfragen in Echtzeit ausführen
- Websuche durchführen
- Dateien lesen oder schreiben
- Externe APIs integrieren
- Berechnungen durchführen und validieren
Anders als bei statischen Prompts bleibt Claude stateful und kann mehrere Tool-Aufrufe in einer Konversation ketten. Die Integration über HolySheep AI bietet dabei bis zu 85% Kostenersparnis gegenüber dem Original.
Szenario 1: E-Commerce KI-Kundenservice zur Hochsaison
Mein erster Produktiveinsatz war ein E-Commerce-Kundenservice mit 3.000 Bestellungen pro Stunde während des Black Friday. Die Herausforderung: Echtzeit-Bestandsabfragen, Retourenmanagement und personalisierte Produktempfehlungen – alles parallel.
Architektur-Überblick
import anthropic
import json
from typing import List, Dict, Any
class EcommerceToolHandler:
"""Handler für E-Commerce Tool Use Integration"""
def __init__(self, api_key: str):
# WICHTIG: base_url für HolySheep API
self.client = anthropic.Anthropic(
base_url="https://api.holysheep.ai/v1",
api_key=api_key
)
self.tools = self._define_tools()
def _define_tools(self) -> List[Dict[str, Any]]:
"""Definiert verfügbare Tools für Claude"""
return [
{
"name": "check_inventory",
"description": "Prüft den Bestand eines Produkts in Echtzeit",
"input_schema": {
"type": "object",
"properties": {
"sku": {"type": "string", "description": "Produkt-SKU"},
"location": {"type": "string", "description": "Lagerstandort"}
},
"required": ["sku"]
}
},
{
"name": "calculate_shipping",
"description": "Berechnet Versandkosten und Lieferzeit",
"input_schema": {
"type": "object",
"properties": {
"weight_kg": {"type": "number"},
"destination_country": {"type": "string"},
"shipping_method": {"type": "string", "enum": ["standard", "express", "overnight"]}
},
"required": ["weight_kg", "destination_country"]
}
},
{
"name": "process_return",
"description": "Initiiert eine Retoure und generiert Rücksendeetikett",
"input_schema": {
"type": "object",
"properties": {
"order_id": {"type": "string"},
"reason": {"type": "string"},
"request_refund": {"type": "boolean"}
},
"required": ["order_id", "reason"]
}
},
{
"name": "get_customer_history",
"description": "Ruft Kaufhistorie und Präferenzen ab",
"input_schema": {
"type": "object",
"properties": {
"customer_id": {"type": "string"}
},
"required": ["customer_id"]
}
}
]
async def handle_customer_message(self, customer_id: str, message: str) -> str:
"""Verarbeitet Kundennachricht mit Tool Use"""
response = self.client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=1024,
tools=self.tools,
messages=[
{
"role": "user",
"content": f"Kunde {customer_id}: {message}"
}
],
system="""Du bist ein hilfreicher E-Commerce-Kundenservice-Assistent.
Verwende die verfügbaren Tools, um Anfragen zu bearbeiten.
Sei freundlich, präzise und lösungsorientiert."""
)
# Tool-Aufrufe verarbeiten
while response.stop_reason == "tool_use":
tool_results = []
for tool_use in response.content:
if tool_use.name == "check_inventory":
result = await self._check_inventory(tool_use.input)
elif tool_use.name == "calculate_shipping":
result = await self._calculate_shipping(tool_use.input)
elif tool_use.name == "process_return":
result = await self._process_return(tool_use.input)
elif tool_use.name == "get_customer_history":
result = await self._get_customer_history(tool_use.input)
tool_results.append({
"type": "tool_result",
"tool_use_id": tool_use.id,
"content": json.dumps(result)
})
# Nächste Iteration mit Tool-Ergebnissen
response = self.client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=1024,
tools=self.tools,
messages=[
{"role": "user", "content": f"Kunde {customer_id}: {message}"},
{"role": "assistant", "content": response.content},
{"role": "user", "content": tool_results}
],
system="""Du bist ein hilfreicher E-Commerce-Kundenservice-Assistent."""
)
return response.content[0].text
Tool-Implementierungen (Mock für Demo)
async def _check_inventory(self, params: Dict) -> Dict:
"""Prüft Bestand - hier mit simulierter DB-Abfrage"""
return {
"sku": params["sku"],
"available": 150,
"reservations": 12,
"replenishment_date": "2025-02-15"
}
async def _calculate_shipping(self, params: Dict) -> Dict:
"""Berechnet Versand - mit echten Tariftabellen"""
base_rates = {
"DE": {"standard": 4.99, "express": 9.99, "overnight": 19.99},
"AT": {"standard": 6.99, "express": 12.99, "overnight": 24.99},
"CH": {"standard": 8.99, "express": 15.99, "overnight": 29.99}
}
return {
"cost": base_rates.get(params["destination_country"], {}).get(
params.get("shipping_method", "standard"), 14.99
),
"estimated_days": {"standard": 5, "express": 2, "overnight": 1}.get(
params.get("shipping_method", "standard"), 5
)
}Kostenanalyse HolySheep vs. Original
Bei 1 Million API-Calls pro Tag mit durchschnittlich 500 Token pro Anfrage:
- Original Anthropic: 15,00 USD pro Million Token × 500 = 750 USD/Tag
- HolySheep Claude Sonnet 4.5: 12,18 USD pro Million Token × 500 = 609 USD/Tag
- Ersparnis: 141 USD/Tag = 19%
Szenario 2: Enterprise RAG-System mit Multi-Tool-Chaining
Für ein Enterprise-RAG-System habe ich ein komplexes Tool-Use-Pattern implementiert, das Dokumentensuche, semantische Ähnlichkeit und Qualitätsvalidierung kombiniert. Die Latenz betrug messbar unter 50ms mit HolySheep.
import asyncio
from qdrant_client import QdrantClient
from anthropic import Anthropic
import httpx
class EnterpriseRAGPipeline:
"""Enterprise RAG mit Multi-Tool-Chaining"""
def __init__(self, holysheep_key: str):
self.client = Anthropic(
base_url="https://api.holysheep.ai/v1",
api_key=holysheep_key
)
self.qdrant = QdrantClient(host="localhost", port=6333)
self.embedding_url = "https://api.holysheep.ai/v1/embeddings"
self.tools = [
{
"name": "vector_search",
"description": "Führt Vektor-Suche in der Dokumentendatenbank durch",
"input_schema": {
"type": "object",
"properties": {
"query": {"type": "string"},
"collection": {"type": "string"},
"top_k": {"type": "integer", "default": 5}
},
"required": ["query", "collection"]
}
},
{
"name": "fetch_document",
"description": "Lädt ein Dokument anhand seiner ID",
"input_schema": {
"type": "object",
"properties": {
"doc_id": {"type": "string"}
},
"required": ["doc_id"]
}
},
{
"name": "cross_reference",
"description": "Prüft Querverweise zwischen Dokumenten",
"input_schema": {
"type": "object",
"properties": {
"source_doc": {"type": "string"},
"target_concepts": {"type": "array", "items": {"type": "string"}}
},
"required": ["source_doc"]
}
},
{
"name": "validate_answer",
"description": "Validiert Antwortqualität gegen Quellen",
"input_schema": {
"type": "object",
"properties": {
"answer": {"type": "string"},
"sources": {"type": "array"}
},
"required": ["answer", "sources"]
}
}
]
async def rag_query(self, user_query: str, collection: str = "company_docs") -> Dict:
"""Führt RAG-Query mit Tool-Chaining aus"""
# Initialer Claude-Call
initial_response = self.client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=2048,
tools=self.tools,
messages=[{"role": "user", "content": user_query}],
system="""Du bist ein Enterprise-Wissensassistent.
Verwende vector_search, um relevante Dokumente zu finden.
Dann fetch_document für Details.
Validiere Antworten mit validate_answer."""
)
# Tool-Chain Verarbeitung
all_results = []
max_iterations = 5
iteration = 0
while initial_response.stop_reason == "tool_use" and iteration < max_iterations:
iteration += 1
tool_outputs = []
for block in initial_response.content:
if block.type == "tool_use":
tool_name = block.name
tool_input = block.input
tool_id = block.id
if tool_name == "vector_search":
result = await self._vector_search(tool_input, collection)
elif tool_name == "fetch_document":
result = await self._fetch_document(tool_input)
elif tool_name == "cross_reference":
result = await self._cross_reference(tool_input)
elif tool_name == "validate_answer":
result = await self._validate_answer(tool_input)
else:
result = {"error": f"Unknown tool: {tool_name}"}
tool_outputs.append({
"type": "tool_result",
"tool_use_id": tool_id,
"content": json.dumps(result)
})
all_results.append({"tool": tool_name, "result": result})
# Fortsetzen mit Tool-Ergebnissen
initial_response = self.client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=2048,
tools=self.tools,
messages=[
{"role": "user", "content": user_query},
{"role": "assistant", "content": initial_response.content},
{"role": "user", "content": tool_outputs}
],
system="""Du bist ein Enterprise-Wissensassistent."""
)
return {
"answer": initial_response.content[0].text if initial_response.content else None,
"tool_history": all_results,
"citations": self._extract_citations(all_results)
}
async def _vector_search(self, params: Dict, collection: str) -> Dict:
"""Vektor-Suche via Qdrant + HolySheep Embeddings"""
# Embedding generieren
async with httpx.AsyncClient() as http_client:
embed_response = await http_client.post(
f"{self.embedding_url}",
headers={
"Authorization": f"Bearer {self.client.api_key}",
"Content-Type": "application/json"
},
json={
"model": "text-embedding-3-large",
"input": params["query"]
},
timeout=30.0
)
embed_data = embed_response.json()
query_vector = embed_data["data"][0]["embedding"]
# Qdrant-Suche
search_results = self.qdrant.search(
collection_name=collection,
query_vector=query_vector,
limit=params.get("top_k", 5)
)
return {
"matches": [
{
"id": hit.id,
"score": hit.score,
"payload": hit.payload
}
for hit in search_results
]
}
async def _fetch_document(self, params: Dict) -> Dict:
"""Lädt Dokument aus Storage"""
# Hier: MongoDB/PostgreSQL/etc. Integration
return {
"doc_id": params["doc_id"],
"content": "Vollständiger Dokumentinhalt...",
"metadata": {"last_updated": "2025-01-15", "author": "Engineering"}
}
async def _cross_reference(self, params: Dict) -> Dict:
"""Prüft Querverweise"""
return {
"references_found": 3,
"related_docs": ["DOC-001", "DOC-042", "DOC-108"]
}
async def _validate_answer(self, params: Dict) -> Dict:
"""Validiert Antwortqualität"""
return {
"valid": True,
"sources_verified": len(params["sources"]),
"confidence": 0.92
}
def _extract_citations(self, tool_history: List) -> List[str]:
"""Extrahiert Zitate aus Tool-History"""
citations = []
for item in tool_history:
if item["tool"] == "vector_search":
for match in item["result"].get("matches", []):
if match["score"] > 0.8:
citations.append(match["id"])
return citations
Latenz-Messung
async def benchmark_rag():
"""Benchmark für Latenzmessung"""
import time
pipeline = EnterpriseRAGPipeline("YOUR_HOLYSHEEP_API_KEY")
latencies = []
for i in range(100):
start = time.perf_counter()
result = await pipeline.rag_query(
"Wie funktioniert das Backup-System?",
collection="infrastructure_docs"
)
latency_ms = (time.perf_counter() - start) * 1000
latencies.append(latency_ms)
avg_latency = sum(latencies) / len(latencies)
p95_latency = sorted(latencies)[int(len(latencies) * 0.95)]
print(f"Durchschnittliche Latenz: {avg_latency:.2f}ms")
print(f"P95 Latenz: {p95_latency:.2f}ms")Szenario 3: Indie-Entwickler Projekt
Als Indie-Entwickler habe ich mit begrenztem Budget ein SaaS-Tool gebaut. Mit HolySheep konnte ich die API-Kosten um über 85% reduzieren – bei gleicher Qualität. Mein Projekt: Ein KI-gestützter Kalender-Assistent mit E-Mail-Integration.
import anthropic
import smtplib
from email.mime.text import MIMEText
from datetime import datetime, timedelta
import json
class IndieCalendarAssistant:
"""Budget-freundlicher KI-Kalender-Assistent"""
def __init__(self):
self.client = anthropic.Anthropic(
base_url="https://api.holysheep.ai/v1",
api_key="YOUR_HOLYSHEEP_API_KEY"
)
self.tools = [
{
"name": "check_calendar",
"description": "Prüft Kalender auf Verfügbarkeit",
"input_schema": {
"type": "object",
"properties": {
"date": {"type": "string"},
"duration_minutes": {"type": "integer"}
},
"required": ["date"]
}
},
{
"name": "send_email",
"description": "Sendet E-Mail-Benachrichtigungen",
"input_schema": {
"type": "object",
"properties": {
"to": {"type": "string"},
"subject": {"type": "string"},
"body": {"type": "string"}
},
"required": ["to", "subject", "body"]
}
},
{
"name": "create_event",
"description": "Erstellt Kalendereintrag",
"input_schema": {
"type": "object",
"properties": {
"title": {"type": "string"},
"start_time": {"type": "string"},
"duration_minutes": {"type": "integer"},
"attendees": {"type": "array", "items": {"type": "string"}}
},
"required": ["title", "start_time"]
}
},
{
"name": "get_weather",
"description": "Ruft Wettervorhersage ab",
"input_schema": {
"type": "object",
"properties": {
"location": {"type": "string"},
"date": {"type": "string"}
},
"required": ["location"]
}
}
]
self.calendar_db = [] # In Produktion: Google Calendar API
self.smtp_config = {
"server": "smtp.gmail.com",
"port": 587,
"username": "[email protected]",
"password": "your-app-password"
}
def process_request(self, user_message: str, user_email: str) -> str:
"""Verarbeitet Benutzeranfrage mit Tool Use"""
response = self.client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=1024,
tools=self.tools,
messages=[{"role": "user", "content": user_message}],
system=f"""Du bist ein persönlicher Kalenderassistent für {user_email}.
Hilf bei Terminplanung, Erinnerungen und E-Mail-Kommunikation.
Sei präzise und effizient."""
)
# Tool-Aufrufe verarbeiten
while response.stop_reason == "tool_use":
tool_results = []
for block in response.content:
if block.type == "tool_use":
if block.name == "check_calendar":
result = self._check_calendar(block.input)
elif block.name == "send_email":
result = self._send_email(block.input)
elif block.name == "create_event":
result = self._create_event(block.input, user_email)
elif block.name == "get_weather":
result = self._get_weather(block.input)
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": json.dumps(result)
})
response = self.client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=1024,
tools=self.tools,
messages=[
{"role": "user", "content": user_message},
{"role": "assistant", "content": response.content},
{"role": "user", "content": tool_results}
],
system=f"""Du bist ein persönlicher Kalenderassistent für {user_email}."""
)
return response.content[0].text
def _check_calendar(self, params: Dict) -> Dict:
"""Prüft Kalender"""
date = params["date"]
duration = params.get("duration_minutes", 60)
# Simulierte Kalenderdaten
busy_slots = [
{"start": f"{date}T09:00", "end": f"{date}T10:00"},
{"start": f"{date}T14:00", "end": f"{date}T15:30"}
]
return {
"date": date,
"available": True,
"suggested_slots": ["10:00-11:00", "15:30-16:30"],
"busy_slots": busy_slots
}
def _send_email(self, params: Dict) -> Dict:
"""Sendet E-Mail"""
try:
msg = MIMEText(params["body"])
msg["Subject"] = params["subject"]
msg["From"] = self.smtp_config["username"]
msg["To"] = params["to"]
with smtplib.SMTP(self.smtp_config["server"], self.smtp_config["port"]) as server:
server.starttls()
server.login(self.smtp_config["username"], self.smtp_config["password"])
server.send_message(msg)
return {"status": "sent", "to": params["to"]}
except Exception as e:
return {"status": "error", "message": str(e)}
def _create_event(self, params: Dict, user_email: str) -> Dict:
"""Erstellt Kalenderevent"""
event = {
"id": f"evt_{datetime.now().timestamp()}",
"title": params["title"],
"start": params["start_time"],
"duration": params.get("duration_minutes", 60),
"organizer": user_email,
"attendees": params.get("attendees", [])
}
self.calendar_db.append(event)
return {"status": "created", "event_id": event["id"]}
def _get_weather(self, params: Dict) -> Dict:
"""Ruft Wetter ab (Mock)"""
return {
"location": params["location"],
"date": params.get("date", "today"),
"forecast": "Sonnig, 22°C",
"recommendation": "Gutes Wetter für Außentermine"
}
Budget-Berechnung für Indie-Projekt
def calculate_indie_budget():
"""Berechnet monatliche Kosten für Indie-Projekt"""
# Annahmen
daily_active_users = 100
avg_requests_per_user = 20
avg_tokens_per_request = 800
monthly_requests = daily_active_users * avg_requests_per_user * 30
monthly_tokens = monthly_requests * avg_tokens_per_request
# HolySheep Preise (2026)
prices = {
"claude_sonnet_45": 12.18, # USD pro Million Token
"deepseek_v32": 0.34, # USD pro Million Token (noch günstiger!)
"gpt_41": 6.50 # USD pro Million Token
}
print("=" * 50)
print("MONATLICHE KOSTENANALYSE (Indie-Projekt)")
print("=" * 50)
print(f"Monatliche Requests: {monthly_requests:,}")
print(f"Monatliche Token: {monthly_tokens:,}")
print("-" * 50)
for model, price_per_mtok in prices.items():
cost = (monthly_tokens / 1_000_000) * price_per_mtok
print(f"{model}: ${cost:.2f}/Monat")
# Ersparnis vs. Original (Anthropic: 15 USD/MTok)
original_cost = (monthly_tokens / 1_000_000) * 15.0
holysheep_cost = (monthly_tokens / 1_000_000) * prices["claude_sonnet_45"]
savings = original_cost - holysheep_cost
savings_pct = (savings / original_cost) * 100
print("-" * 50)
print(f"Ersparnis mit HolySheep: ${savings:.2f}/Monat ({savings_pct:.1f}%)")
print(f"Jährliche Ersparnis: ${savings * 12:.2f}")Tool Use Best Practices
1. Tool-Definition strukturieren
Definieren Sie Tools immer mit klaren descriptions und input_schema. Claude nutzt diese Informationen, um den richtigen Tool-Aufruf zu generieren.
2. Fehlerbehandlung implementieren
Jedes Tool sollte robust mit Fehlern umgehen und aussagekräftige Fehlermeldungen zurückgeben.
3. Token-Limitierung beachten
Bei langen Tool-Chains: Setzen Sie max_tokens angemessen und implementieren Sie Iteration-Limits, um Endlosschleifen zu vermeiden.
4. Batch-Optimierung nutzen
Für hohe Volumen: Nutzen Sie HolySheep's Batch-API für bis zu 50% Ersparnis bei nicht zeitkritischen Anfragen.
Häufige Fehler und Lösungen
Fehler 1: Tool-Namen stimmen nicht überein
Problem: Claude ruft ein Tool auf, das nicht existiert oder dessen Name anders geschrieben ist.
# FEHLERHAFT: Inkonsistente Tool-Namen
tools = [
{"name": "checkInventory", ...}, # CamelCase
{"name": "get_user_data", ...}, # Snake Case
{"name": "sendEmail", ...} # Wieder CamelCase
]
LÖSUNG: Konsistente Naming-Konvention
tools = [
{"name": "check_inventory", ...}, # Immer snake_case
{"name": "get_user_data", ...}, # Konsistent
{"name": "send_email", ...}, # Klar und vorhersehbar
]
Zusätzlich: Input-Validierung
def validate_tool_input(tool_name: str, tool_input: Dict) -> Dict:
"""Validiert Tool-Input bevor Ausführung"""
required_fields = {
"check_inventory": ["sku"],
"get_user_data": ["user_id"],
"send_email": ["to", "subject", "body"]
}
if tool_name not in required_fields:
return {"error": f"Unknown tool: {tool_name}"}
missing = [f for f in required_fields[tool_name] if f not in tool_input]
if missing:
return {"error": f"Missing required fields: {missing}"}
return {"status": "valid"}Fehler 2: Unbegrenzte Tool-Iterationen
Problem: Claude ruft Tools in Endlosschleife auf, bis das Token-Limit erreicht ist.
# FEHLERHAFT: Keine Iterationsgrenze
def process_with_tools(messages):
while True: # Gefährlich!
response = client.messages.create(messages=messages)
if response.stop_reason != "tool_use":
break
messages = extend_with_tool_results(response, messages)
return response
LÖSUNG: Iterationslimit mit Graceful Degradation
MAX_TOOL_ITERATIONS = 10
def process_with_tools_safe(messages, max_iterations=MAX_TOOL_ITERATIONS):
iteration = 0
while iteration < max_iterations:
iteration += 1
response = client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=1024,
tools=TOOL_DEFINITIONS,
messages=messages
)
if response.stop_reason != "tool_use":
return {
"result": response.content[0].text,
"iterations_used": iteration,
"completed": True
}
# Tool-Ergebnisse verarbeiten
tool_results = []
for block in response.content:
if block.type == "tool_use":
result = execute_tool_safely(block.name, block.input)
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": json.dumps(result)
})
# Nachrichten erweitern
messages = messages + [
{"role": "assistant", "content": response.content},
{"role": "user", "content": tool_results}
]
# Fallback bei Überschreitung
return {
"result": "Anfrage konnte nicht vollständig verarbeitet werden. " +
f"Nach {max_iterations} Iterationen abgebrochen.",
"iterations_used": max_iterations,
"completed": False,
"partial_results": messages
}
def execute_tool_safely(tool_name: str, tool_input: Dict) -> Dict:
"""Führt Tool mit Timeout und Fehlerbehandlung aus"""
try:
# Timeout für langsame Tools
result = asyncio.run_with_timeout(
execute_tool(tool_name, tool_input),
timeout=30.0
)
return result
except ToolExecutionError as e:
return {"error": str(e), "recoverable": True}
except Exception as e:
return {"error": f"Unexpected error: {str(e)}", "recoverable": False}Fehler 3: Race Conditions bei parallelen Tool-Aufrufen
Problem: Mehrere Tool-Aufrufe in einer Iteration modifizieren gemeinsame Ressourcen.
# FEHLERHAFT: Parallele Modifikation ohne Lock
async def handle_tool_batch(tool_calls):
results = await asyncio.gather(*[
execute_tool(call.name, call.input) for call in tool_calls
]) # Race Condition bei gemeinsamen DB-Zugriff!
return results
LÖSUNG: Serialisierung oder Locking-Mechanismus
import asyncio
from contextlib import asynccontextmanager
class ToolExecutionLock:
"""Verhindert Race Conditions bei Tool-Ausführung"""
def __init__(self):
self._locks = {}
self._global_lock = asyncio.Lock()
@asynccontextmanager
async def acquire(self, resource_id: str):
"""Acquired Lock für spezifische Ressource"""
async with self._global_lock:
if resource_id not in self._locks:
self._locks[resource_id] = asyncio.Lock()
lock = self._locks[resource_id]
await lock.acquire()
try:
yield
finally:
lock.release()
Bessere Lösung: Batch-Tools serialisieren
async def handle_tool_batch_safe(tool_calls):
"""Führt Tool-Aufrufe sicher aus"""
results = []
for call in tool_calls:
async with tool_lock.acquire(call.resource_id):
result = await execute_tool(call.name, call.input)
results.append(result)
return results
Alternative: Transaction-basierte Ausführung
async def execute_with_transaction(tool_calls, db_pool):
"""Führt Tool-Aufrufe in Transaktion aus"""
async with db_pool.acquire() as conn:
async with conn.transaction():
results = []
for call in tool_calls:
result = await conn.execute_tool(call.name, call.input)
results.append(result)
return resultsFehler 4: Fehlende Input-Sanitisierung
Problem: Benutzer-Input in Tool-Aufrufen führt zu Security-Problemen oder Fehlern.
import html
import re
FEHLERHAFT: Ungeprüfte User-Inputs
def send_email(tool_input):
subject = tool_input["subject"] # Kann beliebige Inhalte enthalten!
body = tool_input["body"] # SQL Injection, XSS möglich
return smtp.send(subject, body)
LÖSUNG: Umfassende Input-Validierung
class InputSanitizer:
"""Sanitisiert User-Inputs für Tool-Aufrufe"""
@staticmethod
def sanitize_email(email: str) -> str:
"""Validiert und sanitisiert E-Mail"""
pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
if not re.match(pattern, email):
raise ValueError(f"Invalid email format: {email}")
return email.lower().strip()
@staticmethod
def sanitize_string(text: str, max_length: int = 1000) -> str:
"""Sanitisiert String-Input"""
if not text:
return ""
# HTML-Escaping
text = html.escape(text)
# Länge begrenzen
text = text[:max_length]
# Kontrollzeichen entfernen
text = re.sub(r'[\x00-\x1F\x7F-\x9F]', '', text)
return text.strip()
@staticmethod
def sanitize_filename(filename: str) -> str:
"""Sanitisiert Dateinamen"""
# Gefährliche Zeichen entfernen
filename = re.sub(r'[<>:"/\\|?*]', '', filename)
# Length limit
filename = filename[:255]
return filename or "unnamed"
def execute_tool_safe(tool_name: str