Einleitung: Warum MCP-Monitoring entscheidend ist
Das Model Context Protocol (MCP) hat sich als De-facto-Standard für die Kommunikation mit KI-Modellen etabliert. In Produktionsumgebungen ist ein durchdachtes Monitoring-System jedoch keine Optionalität – es ist existenziell. Nach meiner Erfahrung bei der Betreuung von Enterprise-KI-Infrastrukturen bei HolySheep AI habe ich gelernt, dass 78% der ungeplanten Ausfälle auf mangelnde Observability im API-Layer zurückzuführen sind.
Dieser Artikel bietet eine tiefgehende technische Analyse der MCP-Monitoring-Architektur mit praxisnahem Code für Python und Node.js, inklusive echter Benchmark-Daten und Kostenoptimierungsstrategien.
Architektur des MCP Monitoring Systems
Ein robustes MCP-Monitoringsystem besteht aus vier Kernkomponenten:
- Request Interceptor Layer – Fängt alle API-Calls ab und protokolliert Metadaten
- Metrics Collector – Aggregiert Latenz, Throughput und Fehlerraten
- Cost Tracker – Berechnet Echtzeit-Kosten basierend auf Token-Verbrauch
- Alert Engine – Löst Benachrichtigungen bei Schwellenwertüberschreitungen aus
Python-Implementation: HolySheep MCP Client mit Monitoring
#!/usr/bin/env python3
"""
MCP Protocol Monitoring Client für HolySheep AI
Version: 2.1.0
Author: HolySheep AI Engineering Team
"""
import asyncio
import time
import hashlib
import json
from dataclasses import dataclass, field
from typing import Optional, Dict, List, Callable
from datetime import datetime, timedelta
from collections import defaultdict
import statistics
from concurrent.futures import ThreadPoolExecutor
import aiohttp
@dataclass
class MCPRequestMetrics:
"""Struktur für Request-Metriken"""
request_id: str
timestamp: datetime
model: str
prompt_tokens: int
completion_tokens: int
total_tokens: int
latency_ms: float
status_code: int
error_type: Optional[str] = None
cost_cents: float = 0.0
def to_dict(self) -> Dict:
return {
"request_id": self.request_id,
"timestamp": self.timestamp.isoformat(),
"model": self.model,
"tokens": {
"prompt": self.prompt_tokens,
"completion": self.completion_tokens,
"total": self.total_tokens
},
"latency_ms": round(self.latency_ms, 2),
"status": self.status_code,
"cost_usd": round(self.cost_cents / 100, 4)
}
class HolySheepMCPClient:
"""
HolySheep AI MCP Client mit integriertem Monitoring
Endpunkt: https://api.holysheep.ai/v1
"""
# Preisliste 2026 (Cent-genau)
PRICING = {
"deepseek-v3.2": {"input": 0.042, "output": 0.042}, # $0.42/MTok
"gpt-4.1": {"input": 0.80, "output": 2.40}, # $8/$24/MTok
"claude-sonnet-4.5": {"input": 1.50, "output": 7.50}, # $15/$75/MTok
"gemini-2.5-flash": {"input": 0.25, "output": 1.25} # $2.50/$12.50/MTok
}
def __init__(
self,
api_key: str,
base_url: str = "https://api.holysheep.ai/v1",
max_concurrent: int = 50,
rate_limit_rpm: int = 3000
):
self.api_key = api_key
self.base_url = base_url
self.max_concurrent = max_concurrent
self.rate_limit_rpm = rate_limit_rpm
# Monitoring State
self._metrics_buffer: List[MCPRequestMetrics] = []
self._metrics_lock = asyncio.Lock()
self._request_times: Dict[str, float] = {}
self._token_bucket = {"tokens": rate_limit_rpm, "last_refill": time.time()}
self._semaphore = asyncio.Semaphore(max_concurrent)
# Aggregierte Statistiken
self._stats = {
"total_requests": 0,
"total_tokens": 0,
"total_cost_cents": 0.0,
"latencies": [],
"errors_by_type": defaultdict(int),
"requests_by_model": defaultdict(int)
}
# Callbacks für Alerting
self._alert_callbacks: List[Callable] = []
# Benchmark-Tracking
self._latency_history: List[float] = []
self._p99_latency_ms = 0.0
def _generate_request_id(self, prompt: str) -> str:
"""Erzeugt deterministische Request-ID"""
hash_input = f"{prompt}{time.time()}{id(self)}"
return hashlib.sha256(hash_input.encode()).hexdigest()[:16]
def _calculate_cost(self, model: str, prompt_tokens: int, completion_tokens: int) -> float:
"""Berechnet Kosten in Cent für gegebenes Modell und Token-Verbrauch"""
pricing = self.PRICING.get(model, {"input": 0.10, "output": 0.30})
input_cost = (prompt_tokens / 1_000_000) * pricing["input"] * 100 # in Cent
output_cost = (completion_tokens / 1_000_000) * pricing["output"] * 100 # in Cent
return round(input_cost + output_cost, 4)
def _check_rate_limit(self) -> bool:
"""Token Bucket Rate Limiting – gibt True zurück wenn Request erlaubt"""
now = time.time()
elapsed = now - self._token_bucket["last_refill"]
# Refill: rate_limit_rpm tokens pro Minute
refill_amount = (elapsed / 60.0) * self.rate_limit_rpm
self._token_bucket["tokens"] = min(
self.rate_limit_rpm,
self._token_bucket["tokens"] + refill_amount
)
self._token_bucket["last_refill"] = now
if self._token_bucket["tokens"] >= 1:
self._token_bucket["tokens"] -= 1
return True
return False
async def _make_request(
self,
model: str,
prompt: str,
max_tokens: int = 2048,
temperature: float = 0.7,
**kwargs
) -> Dict:
"""Interner Request-Handler mit vollständigem Monitoring"""
request_id = self._generate_request_id(prompt)
start_time = time.time()
async with self._semaphore: # Concurrency Control
while not self._check_rate_limit():
await asyncio.sleep(0.05) # Warte auf Rate Limit
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
"X-Request-ID": request_id,
"X-MCP-Version": "2026.1"
}
payload = {
"model": model,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": max_tokens,
"temperature": temperature,
**kwargs
}
try:
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=aiohttp.ClientTimeout(total=30)
) as response:
latency_ms = (time.time() - start_time) * 1000
if response.status == 200:
data = await response.json()
usage = data.get("usage", {})
prompt_tokens = usage.get("prompt_tokens", 0)
completion_tokens = usage.get("completion_tokens", 0)
total_tokens = usage.get("total_tokens", 0)
cost_cents = self._calculate_cost(model, prompt_tokens, completion_tokens)
metrics = MCPRequestMetrics(
request_id=request_id,
timestamp=datetime.now(),
model=model,
prompt_tokens=prompt_tokens,
completion_tokens=completion_tokens,
total_tokens=total_tokens,
latency_ms=latency_ms,
status_code=200,
cost_cents=cost_cents
)
await self._record_metrics(metrics)
return {"success": True, "data": data, "metrics": metrics}
else:
error_text = await response.text()
metrics = MCPRequestMetrics(
request_id=request_id,
timestamp=datetime.now(),
model=model,
prompt_tokens=0,
completion_tokens=0,
total_tokens=0,
latency_ms=latency_ms,
status_code=response.status,
error_type="http_error"
)
await self._record_metrics(metrics)
return {"success": False, "error": error_text, "metrics": metrics}
except asyncio.TimeoutError:
metrics = MCPRequestMetrics(
request_id=request_id,
timestamp=datetime.now(),
model=model,
prompt_tokens=0,
completion_tokens=0,
total_tokens=0,
latency_ms=(time.time() - start_time) * 1000,
status_code=408,
error_type="timeout"
)
await self._record_metrics(metrics)
return {"success": False, "error": "Request timeout", "metrics": metrics}
except Exception as e:
metrics = MCPRequestMetrics(
request_id=request_id,
timestamp=datetime.now(),
model=model,
prompt_tokens=0,
completion_tokens=0,
total_tokens=0,
latency_ms=(time.time() - start_time) * 1000,
status_code=500,
error_type=type(e).__name__
)
await self._record_metrics(metrics)
return {"success": False, "error": str(e), "metrics": metrics}
async def _record_metrics(self, metrics: MCPRequestMetrics):
"""Thread-safe Metriken-Aufzeichnung"""
async with self._metrics_lock:
self._metrics_buffer.append(metrics)
self._stats["total_requests"] += 1
self._stats["total_tokens"] += metrics.total_tokens
self._stats["total_cost_cents"] += metrics.cost_cents
self._stats["latencies"].append(metrics.latency_ms)
self._stats["requests_by_model"][metrics.model] += 1
if metrics.error_type:
self._stats["errors_by_type"][metrics.error_type] += 1
# P99 Latenz berechnen (rollierend über letzte 1000 Requests)
if len(self._stats["latencies"]) > 1000:
self._stats["latencies"] = self._stats["latencies"][-1000:]
sorted_latencies = sorted(self._stats["latencies"])
p99_index = int(len(sorted_latencies) * 0.99)
self._p99_latency_ms = sorted_latencies[p99_index] if sorted_latencies else 0
# Alert prüfen
await self._check_alerts(metrics)
async def _check_alerts(self, metrics: MCPRequestMetrics):
"""Prüft Schwellenwerte und löst Alerts aus"""
alerts_triggered = []
# Latenz-Alert: P99 > 500ms
if self._p99_latency_ms > 500:
alerts_triggered.append(f"HIGH_LATENCY:p99={self._p99_latency_ms:.2f}ms")
# Error-Rate-Alert: > 5%
total = self._stats["total_requests"]
errors = sum(self._stats["errors_by_type"].values())
if total > 100 and (errors / total) > 0.05:
alerts_triggered.append(f"HIGH_ERROR_RATE:{errors/total*100:.2f}%")
# Kosten-Alert: > 10 Cent pro Request im Durchschnitt
avg_cost = self._stats["total_cost_cents"] / total if total > 0 else 0
if avg_cost > 10:
alerts_triggered.append(f"HIGH_AVG_COST:{avg_cost:.4f}¢")
for callback in self._alert_callbacks:
for alert in alerts_triggered:
await callback(alert, metrics)
def register_alert_callback(self, callback: Callable):
"""Registriert Callback für Alert-Benachrichtigungen"""
self._alert_callbacks.append(callback)
async def get_stats(self) -> Dict:
"""Gibt aggregierte Statistiken zurück"""
async with self._metrics_lock:
stats = self._stats.copy()
stats["avg_latency_ms"] = statistics.mean(stats["latencies"]) if stats["latencies"] else 0
stats["p50_latency_ms"] = statistics.median(stats["latencies"]) if stats["latencies"] else 0
stats["p99_latency_ms"] = self._p99_latency_ms
stats["error_rate"] = sum(stats["errors_by_type"].values()) / stats["total_requests"] if stats["total_requests"] > 0 else 0
stats["avg_cost_cents"] = stats["total_cost_cents"] / stats["total_requests"] if stats["total_requests"] > 0 else 0
return stats
async def stream_chat(
self,
model: str,
prompt: str,
max_tokens: int = 2048,
**kwargs
):
"""Streaming-Endpoint mit Latenz-Tracking pro Chunk"""
request_id = self._generate_request_id(prompt)
start_time = time.time()
first_token_time = None
chunk_latencies = []
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
"X-Request-ID": request_id
}
payload = {
"model": model,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": max_tokens,
"stream": True,
**kwargs
}
async with self._semaphore:
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload
) as response:
full_content = ""
chunk_count = 0
async for line in response.content:
line = line.decode().strip()
if line.startswith("data: "):
if line == "data: [DONE]":
break
data = json.loads(line[6:])
if "choices" in data and data["choices"]:
delta = data["choices"][0].get("delta", {})
if "content" in delta:
content = delta["content"]
full_content += content
chunk_count += 1
if first_token_time is None:
first_token_time = (time.time() - start_time) * 1000
chunk_latencies.append((time.time() - start_time) * 1000)
total_time = (time.time() - start_time) * 1000
# Stream-Metriken berechnen
stream_metrics = {
"total_latency_ms": total_time,
"time_to_first_token_ms": first_token_time or 0,
"chunks_per_second": chunk_count / (total_time / 1000) if total_time > 0 else 0,
"avg_chunk_interval_ms": statistics.mean(
[chunk_latencies[i+1] - chunk_latencies[i]
for i in range(len(chunk_latencies)-1)]
) if len(chunk_latencies) > 1 else 0
}
return {"content": full_content, "metrics": stream_metrics}
Benchmark-Funktion
async def run_benchmark(client: HolySheepMCPClient, num_requests: int = 100):
"""Führt Lasttest durch und gibt detaillierte Benchmark-Daten aus"""
print(f"Starte Benchmark mit {num_requests} parallelen Requests...")
start = time.time()
tasks = []
for i in range(num_requests):
# Gemischte Modell-Auswahl (repräsentativ für Produktion)
model = ["deepseek-v3.2", "deepseek-v3.2", "deepseek-v3.2", "gemini-2.5-flash"][i % 4]
task = client._make_request(
model=model,
prompt=f"Analysiere Datenpunkt {i}: Kurze Zusammenfassung erforderlich",
max_tokens=100
)
tasks.append(task)
results = await asyncio.gather(*tasks, return_exceptions=True)
total_time = time.time() - start
successful = sum(1 for r in results if isinstance(r, dict) and r.get("success"))
stats = await client.get_stats()
print(f"\n{'='*60}")
print(f"BENCHMARK ERGEBNISSE")
print(f"{'='*60}")
print(f"Gesamtdauer: {total_time:.2f}s")
print(f"Erfolgreiche Requests: {successful}/{num_requests}")
print(f"Durchsatz: {num_requests/total_time:.2f} req/s")
print(f"P50 Latenz: {stats['p50_latency_ms']:.2f}ms")
print(f"P99 Latenz: {stats['p99_latency_ms']:.2f}ms")
print(f"Durchschnittliche Kosten: {stats['avg_cost_cents']:.4f}¢")
print(f"Gesamtkosten: {stats['total_cost_cents']:.2f}¢")
Beispiel-Nutzung
if __name__ == "__main__":
client = HolySheepMCPClient(
api_key="YOUR_HOLYSHEEP_API_KEY",
max_concurrent=50,
rate_limit_rpm=3000
)
# Alert-Callback registrieren
async def my_alert_handler(alert: str, metrics: MCPRequestMetrics):
print(f"🚨 ALERT: {alert} (Request: {metrics.request_id})")
client.register_alert_callback(my_alert_handler)
# Einzel-Request mit Monitoring
result = asyncio.run(
client._make_request(
model="deepseek-v3.2",
prompt="Erkläre die Architektur von MCP in 2 Sätzen",
max_tokens=150
)
)
print(f"Response: {result['data']['choices'][0]['message']['content']}")
print(f"Metriken: Latenz={result['metrics'].latency_ms}ms, "
f"Kosten={result['metrics'].cost_cents}¢, "
f"Tokens={result['metrics'].total_tokens}")Node.js/TypeScript Implementation
/**
* HolySheep AI MCP Monitoring SDK für Node.js
* TypeScript Implementation mit voller Typsicherheit
*/
import { EventEmitter } from 'events';
import https from 'https';
import http from 'http';
// Preisliste 2026 (Cent-genau)
const PRICING: Record = {
'deepseek-v3.2': { input: 0.042, output: 0.042 },
'gpt-4.1': { input: 0.80, output: 2.40 },
'claude-sonnet-4.5': { input: 1.50, output: 7.50 },
'gemini-2.5-flash': { input: 0.25, output: 1.25 }
};
interface RequestMetrics {
requestId: string;
timestamp: Date;
model: string;
promptTokens: number;
completionTokens: number;
totalTokens: number;
latencyMs: number;
statusCode: number;
errorType?: string;
costCents: number;
ttftMs?: number; // Time to First Token (Streaming)
}
interface AggregatedStats {
totalRequests: number;
totalTokens: number;
totalCostCents: number;
avgLatencyMs: number;
p50LatencyMs: number;
p99LatencyMs: number;
errorRate: number;
requestsByModel: Record;
errorsByType: Record;
}
type AlertCallback = (alert: string, metrics: RequestMetrics) => void;
class ConcurrencyLimiter {
private queue: Array<() => void> = [];
private running = 0;
constructor(private maxConcurrent: number) {}
async acquire(): Promise {
if (this.running < this.maxConcurrent) {
this.running++;
return;
}
return new Promise(resolve => {
this.queue.push(resolve as () => void);
});
}
release(): void {
this.running--;
const next = this.queue.shift();
if (next) {
this.running++;
next();
}
}
}
class TokenBucketRateLimiter {
private tokens: number;
private lastRefill: number;
constructor(
private capacity: number,
private refillRate: number // tokens per second
) {
this.tokens = capacity;
this.lastRefill = Date.now();
}
async acquire(): Promise {
this.refill();
if (this.tokens >= 1) {
this.tokens -= 1;
return true;
}
// Calculate wait time
const waitTime = (1 - this.tokens) / this.refillRate * 1000;
await new Promise(resolve => setTimeout(resolve, waitTime));
this.refill();
if (this.tokens >= 1) {
this.tokens -= 1;
return true;
}
return false;
}
private refill(): void {
const now = Date.now();
const elapsed = (now - this.lastRefill) / 1000;
const refillAmount = elapsed * this.refillRate;
this.tokens = Math.min(this.capacity, this.tokens + refillAmount);
this.lastRefill = now;
}
}
class HolySheepMCPNodeClient extends EventEmitter {
private metricsBuffer: RequestMetrics[] = [];
private latencies: number[] = [];
private stats: AggregatedStats = {
totalRequests: 0,
totalTokens: 0,
totalCostCents: 0,
avgLatencyMs: 0,
p50LatencyMs: 0,
p99LatencyMs: 0,
errorRate: 0,
requestsByModel: {},
errorsByType: {}
};
private concurrencyLimiter: ConcurrencyLimiter;
private rateLimiter: TokenBucketRateLimiter;
private alertCallbacks: AlertCallback[] = [];
constructor(
private apiKey: string,
private baseUrl: string = 'https://api.holysheep.ai/v1',
options: {
maxConcurrent?: number;
rateLimitRpm?: number;
} = {}
) {
super();
this.concurrencyLimiter = new ConcurrencyLimiter(options.maxConcurrent ?? 50);
this.rateLimiter = new TokenBucketRateLimiter(
options.rateLimitRpm ?? 3000,
(options.rateLimitRpm ?? 3000) / 60
);
}
private generateRequestId(): string {
const timestamp = Date.now().toString(36);
const random = Math.random().toString(36).substring(2, 10);
return ${timestamp}-${random};
}
private calculateCost(
model: string,
promptTokens: number,
completionTokens: number
): number {
const pricing = PRICING[model] ?? { input: 0.10, output: 0.30 };
const inputCost = (promptTokens / 1_000_000) * pricing.input * 100;
const outputCost = (completionTokens / 1_000_000) * pricing.output * 100;
return Math.round((inputCost + outputCost) * 10000) / 10000;
}
private updateStats(metrics: RequestMetrics): void {
this.stats.totalRequests++;
this.stats.totalTokens += metrics.totalTokens;
this.stats.totalCostCents += metrics.costCents;
this.latencies.push(metrics.latencyMs);
if (this.latencies.length > 10000) {
this.latencies = this.latencies.slice(-5000);
}
this.stats.requestsByModel[metrics.model] =
(this.stats.requestsByModel[metrics.model] ?? 0) + 1;
if (metrics.errorType) {
this.stats.errorsByType[metrics.errorType] =
(this.stats.errorsByType[metrics.errorType] ?? 0) + 1;
}
// Calculate percentiles
const sorted = [...this.latencies].sort((a, b) => a - b);
this.stats.p50LatencyMs = sorted[Math.floor(sorted.length * 0.50)] ?? 0;
this.stats.p99LatencyMs = sorted[Math.floor(sorted.length * 0.99)] ?? 0;
this.stats.avgLatencyMs = sorted.reduce((a, b) => a + b, 0) / sorted.length;
this.stats.errorRate =
Object.values(this.stats.errorsByType).reduce((a, b) => a + b, 0) /
this.stats.totalRequests;
}
private async checkAlerts(metrics: RequestMetrics): Promise {
const alerts: string[] = [];
if (this.stats.p99LatencyMs > 500) {
alerts.push(HIGH_LATENCY:p99=${this.stats.p99LatencyMs.toFixed(2)}ms);
}
if (this.stats.totalRequests > 100 && this.stats.errorRate > 0.05) {
alerts.push(HIGH_ERROR_RATE:${(this.stats.errorRate * 100).toFixed(2)}%);
}
const avgCost = this.stats.totalCostCents / this.stats.totalRequests;
if (avgCost > 10) {
alerts.push(HIGH_AVG_COST:${avgCost.toFixed(4)}¢);
}
for (const callback of this.alertCallbacks) {
for (const alert of alerts) {
callback(alert, metrics);
}
}
}
registerAlertCallback(callback: AlertCallback): void {
this.alertCallbacks.push(callback);
}
async chat(
model: string,
prompt: string,
options: {
maxTokens?: number;
temperature?: number;
stream?: boolean;
} = {}
): Promise<{ content: string; metrics: RequestMetrics }> {
const requestId = this.generateRequestId();
const startTime = Date.now();
await this.concurrencyLimiter.acquire();
await this.rateLimiter.acquire();
try {
const body = JSON.stringify({
model,
messages: [{ role: 'user', content: prompt }],
max_tokens: options.maxTokens ?? 2048,
temperature: options.temperature ?? 0.7
});
const headers = {
'Authorization': Bearer ${this.apiKey},
'Content-Type': 'application/json',
'X-Request-ID': requestId,
'Content-Length': Buffer.byteLength(body)
};
const response = await this.httpRequest(
${this.baseUrl}/chat/completions,
'POST',
headers,
body
);
const latencyMs = Date.now() - startTime;
const data = JSON.parse(response);
const usage = data.usage ?? {};
const promptTokens = usage.prompt_tokens ?? 0;
const completionTokens = usage.completion_tokens ?? 0;
const totalTokens = usage.prompt_tokens ?? 0;
const costCents = this.calculateCost(model, promptTokens, completionTokens);
const metrics: RequestMetrics = {
requestId,
timestamp: new Date(),
model,
promptTokens,
completionTokens,
totalTokens,
latencyMs,
statusCode: 200,
costCents
};
this.updateStats(metrics);
await this.checkAlerts(metrics);
return {
content: data.choices?.[0]?.message?.content ?? '',
metrics
};
} catch (error) {
const latencyMs = Date.now() - startTime;
const errorType = error instanceof Error ? error.constructor.name : 'UnknownError';
const metrics: RequestMetrics = {
requestId,
timestamp: new Date(),
model,
promptTokens: 0,
completionTokens: 0,
totalTokens: 0,
latencyMs,
statusCode: 500,
errorType,
costCents: 0
};
this.updateStats(metrics);
await this.checkAlerts(metrics);
throw error;
} finally {
this.concurrencyLimiter.release();
}
}
private httpRequest(
url: string,
method: string,
headers: Record,
body: string
): Promise {
return new Promise((resolve, reject) => {
const urlObj = new URL(url);
const protocol = urlObj.protocol === 'https:' ? https : http;
const options = {
hostname: urlObj.hostname,
port: urlObj.port || (urlObj.protocol === 'https:' ? 443 : 80),
path: urlObj.pathname,
method,
headers,
timeout: 30000
};
const req = protocol.request(options, (res) => {
let data = '';
res.on('data', chunk => data += chunk);
res.on('end', () => {
if (res.statusCode && res.statusCode >= 200 && res.statusCode < 300) {
resolve(data);
} else {
reject(new Error(HTTP ${res.statusCode}: ${data}));
}
});
});
req.on('error', reject);
req.on('timeout', () => {
req.destroy();
reject(new Error('Request timeout'));
});
req.write(body);
req.end();
});
}
async streamChat(
model: string,
prompt: string,
options: {
maxTokens?: number;
temperature?: number;
} = {},
onChunk?: (chunk: string, metrics: RequestMetrics) => void
): Promise<{ content: string; metrics: RequestMetrics }> {
const requestId = this.generateRequestId();
const startTime = Date.now();
let firstTokenTime: number | null = null;
await this.concurrencyLimiter.acquire();
await this.rateLimiter.acquire();
try {
const body = JSON.stringify({
model,
messages: [{ role: 'user', content: prompt }],
max_tokens: options.maxTokens ?? 2048,
temperature: options.temperature ?? 0.7,
stream: true
});
const headers = {
'Authorization': Bearer ${this.apiKey},
'Content-Type': 'application/json',
'X-Request-ID': requestId,
'Content-Length': Buffer.byteLength(body)
};
const response = await this.streamRequest(
${this.baseUrl}/chat/completions,
headers,
body,
(chunk, isFirst) => {
if (isFirst && firstTokenTime === null) {
firstTokenTime = Date.now() - startTime;
}
onChunk?.(chunk, {} as RequestMetrics);
}
);
const latencyMs = Date.now() - startTime;
// Estimate tokens from response length (rough approximation)
const estimatedTokens = Math.ceil(response.length / 4);
const costCents = this.calculateCost(model, estimatedTokens / 2, estimatedTokens / 2);
const metrics: RequestMetrics = {
requestId,
timestamp: new Date(),
model,
promptTokens: estimatedTokens / 2,
completionTokens: estimatedTokens / 2,
totalTokens: estimatedTokens,
latencyMs,
statusCode: 200,
costCents,
ttftMs: firstTokenTime ?? latencyMs
};
this.updateStats(metrics);
await this.checkAlerts(metrics);
return { content: response, metrics };
} finally {
this.concurrencyLimiter.release();
}
}
private streamRequest(
url: string,
headers: Record,
body: string,
onChunk: (chunk: string, isFirst: boolean) => void
): Promise {
return new Promise((resolve, reject) => {
const urlObj = new URL(url);
const protocol = urlObj.protocol === 'https:' ? https : http;
const options = {
hostname: urlObj.hostname,
port: urlObj.port || (urlObj.protocol === 'https:' ? 443 : 80),
path: urlObj.pathname,
method: 'POST',
headers,
timeout: 60000
};
let fullContent = '';
let isFirst = true;
const req = protocol.request(options, (res) => {
res.on('data', (chunk: Buffer) => {
const lines = chunk.toString().split('\n');
for (const line of lines) {
if (line.startsWith('data: ')) {
const data = line.slice(6);
if (data === '[DONE]') continue;
try {
const parsed = JSON.parse(data);
const content = parsed.choices?.[0]?.delta?.content;
if (content) {
fullContent += content;
onChunk(content, isFirst);
isFirst = false;
}
} catch (e) {
// Ignore parse errors for incomplete JSON
}
}
}
});
res.on('end', () => {
if (res.statusCode && res