How to Ship a Real Autonomous Agent: MCP Architecture Patterns That Actually Work in Production

sequenceDiagram
    participant User
    participant Agent as Agent Runtime
    participant LLM
    participant MCP_S1 as MCP Server (Database)
    participant MCP_S2 as MCP Server (API)

    User->>Agent: Natural language request
    Agent->>MCP_S1: tools/list
    Agent->>MCP_S2: tools/list
    MCP_S1-->>Agent: Tool schemas
    MCP_S2-->>Agent: Tool schemas
    Agent->>LLM: Context + tools + user input
    LLM-->>Agent: Tool call decision
    Agent->>MCP_S1: tools/call {name, arguments}
    MCP_S1-->>Agent: Result
    Agent->>LLM: Result + updated context
    LLM-->>Agent: Final response
    Agent-->>User: Answer

// Agent runtime skeleton — the core execution loop
import { MCPClient } from '@modelcontextprotocol/sdk/client';
import { StreamableHTTPClientTransport } from '@modelcontextprotocol/sdk/client/streamableHttp';
import { randomUUID } from 'crypto';

interface AgentConfig {
  model: string;
  maxIterations: number;
  contextBudget: number; // tokens
  toolTimeoutMs: number;
  circuitBreakerThreshold: number;
}

interface AgentSession {
  sessionId: string;
  messages: Message[];
  toolRegistry: ToolDefinition[];
  activeTools: string[];
  iteration: number;
  startedAt: Date;
}

class AgentRuntime {
  private mcpClients: Map<string, MCPClient> = new Map();
  private sessionStore: SessionStore;
  private circuitBreakers: Map<string, CircuitBreaker> = new Map();

  async initialize(servers: MCPServerConfig[]): Promise<void> {
    for (const server of servers) {
      const client = new MCPClient({
        capabilities: { tools: {}, resources: {}, prompts: {} }
      });

      const transport = new StreamableHTTPClientTransport(
        new URL(server.endpoint)
      );

      await client.connect(transport);
      this.mcpClients.set(server.id, client);
    }
  }

  async execute(sessionId: string, userMessage: string): Promise<string> {
    const session = await this.sessionStore.load(sessionId);
    const config = this.getConfig();

    session.messages.push({ role: 'user', content: userMessage });

    for (let i = 0; i < config.maxIterations; i++) {
      session.iteration = i;

      const context = this.assembleContext(session, config.contextBudget);
      const toolSchemas = this.getActiveToolSchemas(session);

      const response = await this.llm.generate({
        model: config.model,
        messages: context,
        tools: toolSchemas,
        temperature: 0.2,
      });

      if (response.toolCalls.length === 0) {
        session.messages.push({ role: 'assistant', content: response.content });
        await this.sessionStore.save(session);
        return response.content;
      }

      // Execute tool calls
      const results = await this.executeToolCalls(
        response.toolCalls,
        config.toolTimeoutMs
      );

      session.messages.push({
        role: 'assistant',
        content: response.content,
        toolCalls: response.toolCalls,
      });

      for (const result of results) {
        session.messages.push({
          role: 'tool',
          toolCallId: result.toolCallId,
          content: result.output,
        });
      }
    }

    throw new AgentMaxIterationsError(sessionId, config.maxIterations);
  }
}

┌─────────────────────────────────────────────────────────┐
│                    Agent Runtime Pool                     │
│  (stateless, auto-scaled, load-balanced)                 │
└──────────┬──────────────┬──────────────┬─────────────────┘
           │              │              │
     ┌─────▼─────┐  ┌────▼─────┐  ┌───▼──────┐
     │  MCP: DB  │  │ MCP: API │  │MCP: Files│
     │  Server   │  │ Server   │  │ Server   │
     └───────────┘  └──────────┘  └──────────┘

// The execution loop naturally supports sequential chaining
async executeToolCallsSequential(
  toolCalls: ToolCall[],
  timeoutMs: number
): Promise<ToolResult[]> {
  const results: ToolResult[] = [];

  for (const call of toolCalls) {
    const result = await this.executeWithCircuitBreaker(call, timeoutMs);
    results.push(result);
  }

  return results;
}

async executeToolCallsParallel(
  toolCalls: ToolCall[],
  timeoutMs: number
): Promise<ToolResult[]> {
  // Partition into independent groups
  const groups = this.partitionIndependentCalls(toolCalls);

  const allResults: ToolResult[] = [];

  // Execute groups sequentially, calls within group in parallel
  for (const group of groups) {
    const results = await Promise.allSettled(
      group.map(call => this.executeWithCircuitBreaker(call, timeoutMs))
    );

    for (const [i, settled] of results.entries()) {
      if (settled.status === 'fulfilled') {
        allResults.push(settled.value);
      } else {
        allResults.push(this.buildErrorResult(group[i], settled.reason));
      }
    }
  }

  return allResults;
}

private partitionIndependentCalls(toolCalls: ToolCall[]): ToolCall[][] {
  // Topological sort by declared dependencies
  // Calls without dependencies form the first group
  const dependencyMap = new Map<string, string[]>();

  for (const call of toolCalls) {
    if (call.dependencies) {
      dependencyMap.set(call.id, call.dependencies);
    }
  }

  // Kahn's algorithm for topological grouping
  const groups: ToolCall[][] = [];
  let remaining = [...toolCalls];

  while (remaining.length > 0) {
    const ready = remaining.filter(call => {
      const deps = dependencyMap.get(call.id) ?? [];
      return deps.every(dep =>
        groups.flat().some(completed => completed.id === dep)
      );
    });

    if (ready.length === 0) {
      // Circular dependency — fail fast
      throw new CircularDependencyError(toolCalls);
    }

    groups.push(ready);
    remaining = remaining.filter(c => !ready.includes(c));
  }

  return groups;
}

interface IntentRouter {
  route(userMessage: string): Promise<RouteDecision>;
}

interface RouteDecision {
  activeServerIds: string[];
  intent: string;
  confidence: number;
  fallbackServers: string[];
}

class LightweightRouter implements IntentRouter {
  async route(userMessage: string): Promise<RouteDecision> {
    // Use a small, fast model for classification (not your main agent model)
    const classification = await this.smallModel.classify({
      model: 'claude-3-haiku',
      messages: [{
        role: 'user',
        content: `Classify this request into one of: database, api, files, search, compute\n\nRequest: ${userMessage}`
      }],
      maxTokens: 50,
    });

    const intent = this.parseIntent(classification);
    const serverMap: Record<string, string[]> = {
      database: ['mcp-postgres', 'mcp-redis'],
      api: ['mcp-weather', 'mcp-stripe'],
      files: ['mcp-gcs', 'mcp-s3'],
      search: ['mcp-vector', 'mcp-elastic'],
      compute: ['mcp-python', 'mcp-shell'],
    };

    return {
      activeServerIds: serverMap[intent] ?? [],
      intent,
      confidence: this.computeConfidence(classification),
      fallbackServers: ['mcp-general'],
    };
  }
}

// In the agent runtime, before the main loop:
async execute(sessionId: string, userMessage: string): Promise<string> {
  const session = await this.sessionStore.load(sessionId);
  const route = await this.intentRouter.route(userMessage);

  // Activate only relevant tools — reduces schema tokens by 60-80%
  session.activeTools = route.activeServerIds;

  if (route.confidence < 0.6) {
    // Low confidence — include fallback tools
    session.activeTools.push(...route.fallbackServers);
  }

  // ... proceed with main loop using only active tools
}

interface ContextBudget {
  totalWindow: number;          // e.g., 200,000 for Claude 3.5 Sonnet
  systemPrompt: number;         // Fixed overhead
  toolSchemas: number;          // Dynamic — varies by active tools
  conversationHistory: number;  // Grows with each turn
  toolResults: number;          // Grows with each tool call
  outputReserve: number;        // Always reserve for the response
}

function calculateBudget(
  window: number,
  activeTools: ToolDefinition[],
  historyMessages: Message[]
): ContextBudget {
  const toolSchemaTokens = activeTools.reduce(
    (sum, tool) => sum + estimateTokens(tool.jsonSchema), 0
  );

  const historyTokens = historyMessages.reduce(
    (sum, msg) => sum + estimateTokens(msg.content), 0
  );

  return {
    totalWindow: window,
    systemPrompt: 500,
    toolSchemas: toolSchemaTokens,
    conversationHistory: historyTokens,
    toolResults: 0, // Computed per iteration
    outputReserve: Math.max(4096, window * 0.05), // 5% minimum
  };
}

class ContextManager {
  private compactionThreshold = 0.85; // Compact at 85% usage

  async prepareContext(
    session: AgentSession,
    budget: ContextBudget
  ): Promise<PreparedContext> {
    const currentUsage = this.calculateUsage(session, budget);

    if (currentUsage / budget.totalWindow > this.compactionThreshold) {
      await this.compact(session);
    }

    return {
      systemPrompt: this.buildSystemPrompt(session),
      messages: session.messages,
      tools: this.getActiveToolSchemas(session),
    };
  }

  private async compact(session: AgentSession): Promise<void> {
    // Strategy 1: Summarize old messages
    const cutoff = session.messages.length - this.keepRecentCount;

    if (cutoff > 0) {
      const oldMessages = session.messages.slice(0, cutoff);
      const summary = await this.llm.summarize({
        model: 'claude-3-haiku',
        messages: [{
          role: 'user',
          content: `Summarize this conversation preserving all key facts, decisions, and tool results:\n\n${oldMessages.map(m => `[${m.role}] ${m.content}`).join('\n')}`
        }],
        maxTokens: 1000,
      });

      session.messages = [
        { role: 'system', content: `[COMPACTED HISTORY]\n${summary}` },
        ...session.messages.slice(cutoff),
      ];
    }

    // Strategy 2: Truncate large tool results
    for (const msg of session.messages) {
      if (msg.role === 'tool' && this.estimateTokens(msg.content) > 2000) {
        msg.content = this.truncateWithSummary(msg.content, 2000);
        msg.truncated = true;
      }
    }

    // Strategy 3: Prune inactive tool schemas
    session.activeTools = session.activeTools.filter(toolId =>
      this.recentlyUsed(toolId, session.messages, 3)
    );
  }
}

class CircuitBreaker {
  private state: 'CLOSED' | 'OPEN' | 'HALF_OPEN' = 'CLOSED';
  private failureCount = 0;
  private lastFailureTime = 0;
  private successCount = 0;

  constructor(
    private serverId: string,
    private threshold: number = 5,
    private resetTimeoutMs: number = 30_000,
    private halfOpenMax: number = 3
  ) {}

  async execute<T>(fn: () => Promise<T>): Promise<T> {
    if (this.state === 'OPEN') {
      if (Date.now() - this.lastFailureTime > this.resetTimeoutMs) {
        this.state = 'HALF_OPEN';
        this.successCount = 0;
      } else {
        throw new CircuitOpenError(this.serverId);
      }
    }

    try {
      const result = await fn();
      this.onSuccess();
      return result;
    } catch (error) {
      this.onFailure(error);
      throw error;
    }
  }

  private onSuccess(): void {
    this.failureCount = 0;
    if (this.state === 'HALF_OPEN') {
      this.successCount++;
      if (this.successCount >= this.halfOpenMax) {
        this.state = 'CLOSED';
      }
    }
  }

  private onFailure(error: Error): void {
    this.failureCount++;
    this.lastFailureTime = Date.now();

    if (this.failureCount >= this.threshold) {
      this.state = 'OPEN';
    }

    // Classify for retry decisions
    const isRetryable = this.isRetryableError(error);
    if (!isRetryable && this.state !== 'OPEN') {
      // Non-retryable — open immediately
      this.failureCount = this.threshold;
      this.state = 'OPEN';
    }
  }

  private isRetryableError(error: Error): boolean {
    // Rate limits, timeouts, 5xx — retryable
    // 4xx (auth, validation), schema errors — NOT retryable
    if (error instanceof RateLimitError) return true;
    if (error instanceof TimeoutError) return true;
    if (error instanceof ServerError && error.status >= 500) return true;
    if (error instanceof AuthError) return false;
    if (error instanceof ValidationError) return false;
    return false;
  }
}

class RetryPolicy {
  constructor(
    private maxRetries: number = 3,
    private baseDelayMs: number = 1000,
    private maxDelayMs: number = 30_000
  ) {}

  async executeWithRetry<T>(
    fn: () => Promise<T>,
    context: RetryContext
  ): Promise<T> {
    let lastError: Error;

    for (let attempt = 0; attempt <= this.maxRetries; attempt++) {
      try {
        return await fn();
      } catch (error) {
        lastError = error as Error;

        if (!this.isRetryable(error) || attempt === this.maxRetries) {
          break;
        }

        const delay = this.calculateDelay(attempt);
        context.onRetry({
          attempt: attempt + 1,
          delay,
          error: lastError,
          toolCallId: context.toolCallId,
        });

        await sleep(delay);
      }
    }

    throw lastError;
  }

  private calculateDelay(attempt: number): number {
    // Exponential backoff with full jitter
    const exponential = Math.min(
      this.maxDelayMs,
      this.baseDelayMs * Math.pow(2, attempt)
    );
    return Math.random() * exponential; // Full jitter
  }
}

private buildErrorResult(
  toolCall: ToolCall,
  error: unknown
): ToolResult {
  const errorMessage = error instanceof Error ? error.message : String(error);

  return {
    toolCallId: toolCall.id,
    output: JSON.stringify({
      error: true,
      tool: toolCall.name,
      message: errorMessage,
      recoverable: this.isRecoverable(error),
      suggestion: this.getRecoverySuggestion(toolCall, error),
    }),
    isError: true,
  };
}

private getRecoverySuggestion(toolCall: ToolCall, error: unknown): string {
  if (error instanceof ValidationError) {
    return `The arguments for '${toolCall.name}' were invalid. ` +
           `Check the schema and try again with corrected arguments.`;
  }
  if (error instanceof RateLimitError) {
    return `The '${toolCall.name}' service is rate-limited. ` +
           `Try a different approach or wait before retrying.`;
  }
  if (error instanceof TimeoutError) {
    return `The '${toolCall.name}' call timed out. ` +
           `Consider breaking the operation into smaller steps.`;
  }
  return `The '${toolCall.name}' call failed. Try an alternative approach.`;
}

interface ToolAuthorization {
  agentId: string;
  userId: string;
  scope: ToolScope[];
  rateLimit: RateLimitPolicy;
  dataClassification: DataLevel;
  expiresAt: Date;
}

interface ToolScope {
  toolName: string;
  permissions: ('read' | 'write' | 'delete')[];
  resourceFilter?: Record<string, unknown>; // e.g., { orgId: 'acme' }
}

// Example: Agent can only read/write its own user's data
const agentAuth: ToolAuthorization = {
  agentId: 'agent-7x2k',
  userId: 'user-123',
  scope: [
    { toolName: 'db.query', permissions: ['read'], resourceFilter: { userId: 'user-123' } },
    { toolName: 'db.insert', permissions: ['write'], resourceFilter: { userId: 'user-123' } },
    { toolName: 'api.weather', permissions: ['read'] },
  ],
  rateLimit: { requestsPerMinute: 60, tokensPerHour: 100_000 },
  dataClassification: 'internal',
  expiresAt: new Date(Date.now() + 30 * 60 * 1000), // 30 min TTL
};

class SandboxedExecutor {
  private sandboxPool: SandboxInstance[];

  async execute(toolCall: ToolCall, auth: ToolAuthorization): Promise<ToolResult> {
    const sandbox = await this.sandboxPool.acquire();

    try {
      // Apply resource limits
      const result = await sandbox.execute({
        code: toolCall.arguments.code,
        timeoutMs: 30_000,
        memoryLimitMb: 256,
        cpuLimit: 0.5, // 50% of one core
        networkAccess: auth.scope.some(s => s.toolName.startsWith('api.')),
        filesystemAccess: auth.scope.some(s => s.toolName.startsWith('file.')),
        environment: {
          // Inject only whitelisted env vars
          OPENAI_API_KEY: this.secretStore.get('OPENAI_API_KEY'),
        },
      });

      return { toolCallId: toolCall.id, output: result.stdout };
    } finally {
      await this.sandboxPool.release(sandbox);
    }
  }
}

class SecurityFilter {
  async sanitizeToolOutput(output: string, source: string): Promise<string> {
    // Remove potential prompt injection patterns from tool results
    const sanitized = output
      .replace(/<\|.*?\|>/g, '[REDACTED]') // LLM system tokens
      .replace(/\b(system|developer)\b\s*:/gi, '[FILTERED]')
      .replace(/ignore\s+(all\s+)?(previous|above)\s+instructions/gi, '[FILTERED]');

    // Truncate to prevent context flooding
    if (this.estimateTokens(sanitized) > 4000) {
      return sanitized.slice(0, this.findSentenceBoundary(sanitized, 4000)) +
             '\n... [output truncated]';
    }

    return sanitized;
  }

  async validateAgentOutput(output: string, policy: OutputPolicy): Promise<string> {
    // Check for data exfiltration patterns
    const sensitivePatterns = [
      /\b\d{4}[-\s]?\d{4}[-\s]?\d{4}[-\s]?\d{4}\b/g, // Credit cards
      /\b\d{3}[-\s]?\d{2}[-\s]?\d{3}[-\s]?\d{4}\b/g, // SSNs
    ];

    for (const pattern of sensitivePatterns) {
      const matches = output.match(pattern);
      if (matches && !policy.allowSensitiveData) {
        throw new OutputPolicyViolation(
          `Output contains ${matches.length} potential sensitive data matches`
        );
      }
    }

    return output;
  }
}

import { trace, span, SpanStatus } from '@opentelemetry/api';

const tracer = trace.getTracer('agent-runtime', '1.0.0');

async execute(sessionId: string, userMessage: string): Promise<string> {
  const rootSpan = tracer.startSpan('agent.execute', {
    attributes: {
      'agent.session_id': sessionId,
      'agent.model': this.config.model,
      'agent.max_iterations': this.config.maxIterations,
    },
  });

  try {
    const session = await this.sessionStore.load(sessionId);

    for (let i = 0; i < this.config.maxIterations; i++) {
      const iterationSpan = tracer.startSpan(`agent.iteration.${i}`, {
        attributes: {
          'agent.iteration': i,
          'agent.context_tokens': this.estimateContextTokens(session),
          'agent.active_tools': session.activeTools.length,
        },
      });

      // Trace LLM call
      const llmSpan = tracer.startSpan('llm.generate', {
        attributes: {
          'llm.model': this.config.model,
          'llm.input_tokens': this.estimateInputTokens(session),
        },
      });

      const response = await this.llm.generate(/* ... */);
      llmSpan.setAttributes({
        'llm.output_tokens': response.usage?.outputTokens ?? 0,
        'llm.tool_calls': response.toolCalls.length,
        'llm.finish_reason': response.finishReason,
      });
      llmSpan.end();

      // Trace each tool call
      for (const toolCall of response.toolCalls) {
        const toolSpan = tracer.startSpan(`tool.${toolCall.name}`, {
          attributes: {
            'tool.name': toolCall.name,
            'tool.server': this.getServerForTool(toolCall.name),
            'tool.arguments_size': JSON.stringify(toolCall.arguments).length,
          },
        });

        try {
          const result = await this.executeWithCircuitBreaker(toolCall);
          toolSpan.setAttributes({
            'tool.status': 'success',
            'tool.result_size': result.output.length,
          });
        } catch (error) {
          toolSpan.setStatus(SpanStatus.ERROR, String(error));
          toolSpan.setAttributes({
            'tool.status': 'error',
            'tool.error_type': error.constructor.name,
          });
        } finally {
          toolSpan.end();
        }
      }

      iterationSpan.end();
    }

    rootSpan.setStatus(SpanStatus.OK);
    return response.content;
  } catch (error) {
    rootSpan.setStatus(SpanStatus.ERROR, String(error));
    throw error;
  } finally {
    rootSpan.end();
  }
}

interface AgentLogEvent {
  timestamp: string;
  sessionId: string;
  iteration: number;
  event: 'tool_call' | 'tool_result' | 'llm_response' | 'error' | 'compaction';
  data: Record<string, unknown>;
}

// Every meaningful event gets logged — this is your debugging lifeline
private logEvent(sessionId: string, event: AgentLogEvent): void {
  this.logger.info(JSON.stringify({
    level: 'info',
    event: event.event,
    session: sessionId,
    iteration: event.iteration,
    ...event.data,
    timestamp: new Date().toISOString(),
  }));
}

class ExecutionRecorder {
  async record(
    sessionId: string,
    iteration: number,
    request: LLMRequest,
    response: LLMResponse
  ): Promise<void> {
    await this.store.put(`replay/${sessionId}/${iteration}`, {
      request: {
        model: request.model,
        messages: request.messages,
        tools: request.tools,
        temperature: request.temperature,
      },
      response: {
        content: response.content,
        toolCalls: response.toolCalls,
        usage: response.usage,
        finishReason: response.finishReason,
      },
      metadata: {
        timestamp: new Date().toISOString(),
        latencyMs: response.latency,
        contextTokens: request.messages.reduce((s, m) => s + m.tokens, 0),
      },
    });
  }
}

// In test/debug mode, swap the real LLM for a replay provider
class ReplayLLMProvider implements LLMProvider {
  constructor(private recorder: ExecutionRecorder) {}

  async generate(request: LLMRequest): Promise<LLMResponse> {
    // Find the matching recorded response
    const recorded = await this.recorder.findMatch(request);
    if (!recorded) {
      throw new Error('No replay data for this request — run in live mode first');
    }
    return recorded;
  }
}

interface AgentTopology {
  orchestrator: OrchestratorConfig;
  workers: WorkerConfig[];
}

interface WorkerConfig {
  id: string;
  name: string;
  mcpServers: string[];
  maxIterations: number;
  model: string;
  contextBudget: number;
}

class OrchestratorAgent {
  async execute(userMessage: string): Promise<string> {
    // Step 1: Plan — decompose into subtasks
    const plan = await this.plan(userMessage);

    // Step 2: Dispatch — send subtasks to specialized workers
    const results = await Promise.all(
      plan.subtasks.map(async (subtask) => {
        const worker = this.workers.get(subtask.workerId);
        return worker.execute(subtask.instructions, subtask.context);
      })
    );

    // Step 3: Synthesize — combine results
    return this.synthesize(userMessage, plan, results);
  }

  private async plan(userMessage: string): Promise<ExecutionPlan> {
    return this.llm.generateStructured<ExecutionPlan>({
      model: 'claude-3-opus',
      systemPrompt: `You are a task planner. Decompose the request into parallel subtasks.
Each subtask must specify which worker agent should handle it.
Available workers: ${this.getWorkerDescriptions()}`,
      userMessage,
      schema: ExecutionPlanSchema,
    });
  }
}

// Worker agent exposes itself as an MCP server
import { MCPServer } from '@modelcontextprotocol/sdk/server';

class AgentAsMCPServer {
  private server: MCPServer;

  constructor(private agent: WorkerAgent) {
    this.server = new MCPServer({
      name: agent.config.name,
      version: '1.0.0',
    });

    // Expose the agent's capabilities as MCP tools
    this.server.registerTool({
      name: `${agent.config.name}.execute`,
      description: `Execute a task using the ${agent.config.name} agent`,
      inputSchema: {
        type: 'object',
        properties: {
          task: { type: 'string', description: 'The task to execute' },
          context: { type: 'string', description: 'Additional context' },
          maxSteps: { type: 'number', description: 'Maximum execution steps' },
        },
        required: ['task'],
      },
      handler: async (args) => {
        const result = await this.agent.execute({
          userMessage: args.task,
          additionalContext: args.context,
          maxIterations: args.maxSteps ?? 5,
        });
        return { content: [{ type: 'text', text: result }] };
      },
    });
  }

  async start(port: number): Promise<void> {
    await this.server.listen({ port, host: '0.0.0.0' });
  }
}

interface ModelRouter {
  route(task: RoutingCriteria): string;
}

class CostAwareModelRouter implements ModelRouter {
  route(criteria: RoutingCriteria): string {
    if (criteria.isClassification) return 'claude-3-haiku';       // $0.25/MTok
    if (criteria.isSummarization) return 'claude-3-haiku';        // $0.25/MTok
    if (criteria.complexity === 'high') return 'claude-3-opus';   // $15/MTok
    if (criteria.complexity === 'medium') return 'claude-3-sonnet'; // $3/MTok
    return 'claude-3-haiku';                                      // Default cheap
  }
}

// In practice: a router agent uses haiku, the main agent uses sonnet,
// and only complex reasoning steps escalate to opus.
// This typically reduces costs by 60-80% vs. using one model for everything.

# mcp_lifecycle.py
import asyncio
from contextlib import asynccontextmanager
from typing import AsyncGenerator
from mcp import ClientSession, StdioServerParameters

class MCPServerPool:
    """
    Manages MCP server instances with automatic lifecycle handling.
    Each session gets an isolated server instance to prevent state leakage.
    """
    
    def __init__(self, max_instances: int = 10, idle_timeout: float = 300.0):
        self.max_instances = max_instances
        self.idle_timeout = idle_timeout
        self._semaphore = asyncio.Semaphore(max_instances)
        self._instances: dict[str, ClientSession] = {}
        self._lock = asyncio.Lock()
    
    @asynccontextmanager
    async def acquire(self, server_params: StdioServerParameters) -> AsyncGenerator[ClientSession, None]:
        """Acquire a session-scoped MCP server instance."""
        async with self._semaphore:
            session_id = str(uuid4())
            session = None
            try:
                session = await self._create_session(server_params)
                async with session:
                    yield session
            except Exception as e:
                logger.error(f"MCP server error for session {session_id}: {e}")
                raise
            finally:
                await self._cleanup(session_id)
    
    async def _create_session(self, params: StdioServerParameters) -> ClientSession:
        """Spawn a fresh MCP server process."""
        from mcp import stdio_client
        
        async with stdio_client(params) as (read_stream, write_stream):
            session = ClientSession(read_stream, write_stream)
            await session.initialize()
            return session
    
    async def _cleanup(self, session_id: str):
        """Ensure resources are released."""
        async with self._lock:
            self._instances.pop(session_id, None)

# mcp_health.py
import time
from dataclasses import dataclass, field
from enum import Enum

class ServerHealth(Enum):
    HEALTHY = "healthy"
    DEGRADED = "degraded"
    UNRESPONSIVE = "unresponsive"
    DEAD = "dead"

@dataclass
class ServerHealthState:
    last_heartbeat: float = field(default_factory=time.time)
    consecutive_failures: int = 0
    total_requests: int = 0
    health: ServerHealth = ServerHealth.HEALTHY
    
    def record_success(self):
        self.consecutive_failures = 0
        self.health = ServerHealth.HEALTHY
        self.last_heartbeat = time.time()
        self.total_requests += 1
    
    def record_failure(self):
        self.consecutive_failures += 1
        self.total_requests += 1
        if self.consecutive_failures >= 3:
            self.health = ServerHealth.UNRESPONSIVE
        elif self.consecutive_failures >= 1:
            self.health = ServerHealth.DEGRADED

class ResilientMCPServer:
    """Wraps an MCP server with automatic health monitoring and recovery."""
    
    def __init__(self, params: StdioServerParameters, circuit_breaker_threshold: int = 5):
        self.params = params
        self.health_state = ServerHealthState()
        self.circuit_breaker_threshold = circuit_breaker_threshold
        self.circuit_open_until: float = 0
        self._session: ClientSession | None = None
    
    async def execute_tool(self, tool_name: str, arguments: dict) -> str:
        """Execute a tool with circuit breaker protection."""
        now = time.time()
        
        # Check circuit breaker
        if now < self.circuit_open_until:
            raise MCPCircuitOpenError(
                f"Circuit breaker open, retry after {self.circuit_open_until - now:.1f}s"
            )
        
        # Check for stale connection
        if self._is_stale():
            await self._reconnect()
        
        try:
            result = await asyncio.wait_for(
                self._session.call_tool(tool_name, arguments),
                timeout=30.0
            )
            self.health_state.record_success()
            return result.content[0].text
        except (asyncio.TimeoutError, ConnectionError) as e:
            self.health_state.record_failure()
            if self.health_state.consecutive_failures >= self.circuit_breaker_threshold:
                self.circuit_open_until = time.time() + 60  # Open for 60 seconds
                raise MCPCircuitOpenError(str(e))
            raise
    
    def _is_stale(self) -> bool:
        return (time.time() - self.health_state.last_heartbeat) > 120
    
    async def _reconnect(self):
        """Tear down and recreate the session."""
        if self._session:
            try:
                await self._session.close()
            except Exception:
                pass
        # Reconnection logic follows the same pattern as initial setup

# mcp_observability.py
import structlog
from dataclasses import dataclass
from datetime import datetime
from uuid import UUID

logger = structlog.get_logger()

@dataclass
class MCPCallEvent:
    session_id: str
    tool_name: str
    arguments: dict
    result: str | None
    error: str | None
    duration_ms: float
    timestamp: datetime
    model_token_usage: int = 0

class MCPTracer:
    """Provides distributed tracing for MCP interactions."""
    
    def __init__(self, service_name: str = "mcp-agent"):
        self.service_name = service_name
        self._traces: list[MCPCallEvent] = []
    
    async def traced_call(self, session: ClientSession, tool_name: str, 
                          arguments: dict, session_id: str):
        """Wrap an MCP tool call with full observability."""
        start_time = time.perf_counter()
        trace_id = str(uuid4())
        
        logger.info("mcp.call.start", 
                    tool=tool_name, 
                    session_id=session_id,
                    trace_id=trace_id,
                    arguments_keys=list(arguments.keys()))
        
        try:
            result = await session.call_tool(tool_name, arguments)
            duration_ms = (time.perf_counter() - start_time) * 1000
            
            event = MCPCallEvent(
                session_id=session_id,
                tool_name=tool_name,
                arguments=arguments,
                result=result.content[0].text,
                error=None,
                duration_ms=duration_ms,
                timestamp=datetime.utcnow()
            )
            self._traces.append(event)
            
            logger.info("mcp.call.success",
                        tool=tool_name,
                        trace_id=trace_id,
                        duration_ms=round(duration_ms, 2),
                        result_length=len(result.content[0].text))
            
            return result
            
        except Exception as e:
            duration_ms = (time.perf_counter() - start_time) * 1000
            
            event = MCPCallEvent(
                session_id=session_id,
                tool_name=tool_name,
                arguments=arguments,
                result=None,
                error=str(e),
                duration_ms=duration_ms,
                timestamp=datetime.utcnow()
            )
            self._traces.append(event)
            
            logger.error("mcp.call.failed",
                         tool=tool_name,
                         trace_id=trace_id,
                         duration_ms=round(duration_ms, 2),
                         error=str(e))
            raise
    
    def get_p99_latency(self, tool_name: str | None = None) -> float:
        """Calculate p99 latency for a specific tool or all tools."""
        events = [e for e in self._traces 
                  if e.error is None and (tool_name is None or e.tool_name == tool_name)]
        if not events:
            return 0.0
        sorted_latencies = sorted(e.duration_ms for e in events)
        p99_index = int(len(sorted_latencies) * 0.99)
        return sorted_latencies[p99_index]

# mcp_metrics.py
from prometheus_client import Counter, Histogram, Gauge

# Tool call counters
mcp_tool_calls_total = Counter(
    'mcp_tool_calls_total',
    'Total MCP tool calls',
    ['tool_name', 'status']  # status: success, failure, timeout
)

# Latency histograms
mcp_tool_latency_seconds = Histogram(
    'mcp_tool_latency_seconds',
    'MCP tool call latency',
    ['tool_name'],
    buckets=[0.1, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0]
)

# Connection health
mcp_server_connections_active = Gauge(
    'mcp_server_connections_active',
    'Number of active MCP server connections'
)

mcp_circuit_breaker_open = Gauge(
    'mcp_circuit_breaker_open',
    'Circuit breaker status (1=open, 0=closed)',
    ['server_name']
)

# Token efficiency
mcp_tokens_per_successful_call = Histogram(
    'mcp_tokens_per_successful_call',
    'LLM tokens consumed per successful MCP tool call',
    ['tool_name'],
    buckets=[50, 100, 200, 500, 1000, 2000, 5000]
)

# mcp_security.py
from pydantic import BaseModel, field_validator
from typing import Any

class ToolPermissionPolicy:
    """Defines what tools an agent can access and with what constraints."""
    
    def __init__(self):
        self._allowed_tools: dict[str, ToolPolicy] = {}
    
    def register_tool(self, tool_name: str, policy: ToolPolicy):
        self._allowed_tools[tool_name] = policy
    
    def validate_call(self, tool_name: str, arguments: dict) -> dict:
        """Validate and sanitize tool arguments before execution."""
        if tool_name not in self._allowed_tools:
            raise ToolPermissionDenied(f"Tool '{tool_name}' is not in the allowlist")
        
        policy = self._allowed_tools[tool_name]
        validated = policy.schema.model_validate(arguments)
        return validated.model_dump()
    
    def get_allowed_tools(self) -> list[str]:
        return list(self._allowed_tools.keys())

class ToolPolicy(BaseModel):
    schema: type[BaseModel]
    max_arguments_size: int = 10_000
    requires_approval: bool = False
    approval_timeout: float = 300.0

# Example: A file system tool with strict validation
class ReadFileArgs(BaseModel):
    path: str
    
    @field_validator("path")
    def validate_path(cls, v: str) -> str:
        if ".." in v:
            raise ValueError("Path traversal not allowed")
        if not v.startswith("/allowed/base/path"):
            raise ValueError("Path must be within allowed directory")
        return v

class WriteFileArgs(BaseModel):
    path: str
    content: str
    
    @field_validator("path")
    def validate_path(cls, v: str) -> str:
        if ".." in v:
            raise ValueError("Path traversal not allowed")
        if not v.startswith("/allowed/base/path"):
            raise ValueError("Path must be within allowed directory")
        return v
    
    @field_validator("content")
    def validate_content_size(cls, v: str) -> str:
        if len(v) > 1_000_000:
            raise ValueError("Content exceeds 1MB limit")
        return v

# mcp_approval.py
import asyncio
from dataclasses import dataclass
from enum import Enum
from typing import Callable, Awaitable

class ApprovalDecision(Enum):
    APPROVED = "approved"
    DENIED = "denied"
    TIMEOUT = "timeout"

@dataclass
class ApprovalRequest:
    tool_name: str
    arguments: dict
    risk_level: str  # "low", "medium", "high", "critical"
    description: str
    agent_id: str
    session_id: str

class ApprovalGate:
    """Blocks agent execution until a human approves sensitive operations."""
    
    def __init__(self, approval_callback: Callable[[ApprovalRequest], Awaitable[ApprovalDecision]]):
        self._callback = approval_callback
        self._pending: dict[str, asyncio.Future[ApprovalDecision]] = {}
    
    async def check(self, tool_name: str, arguments: dict, 
                    risk_level: str, description: str,
                    agent_id: str, session_id: str) -> bool:
        """Check if a tool call requires approval and obtain it if needed."""
        if risk_level not in ("high", "critical"):
            return True
        
        request = ApprovalRequest(
            tool_name=tool_name,
            arguments=arguments,
            risk_level=risk_level,
            description=description,
            agent_id=agent_id,
            session_id=session_id
        )
        
        loop = asyncio.get_event_loop()
        future = loop.create_future()
        request_id = str(uuid4())
        self._pending[request_id] = future
        
        try:
            decision = await asyncio.wait_for(future, timeout=300.0)
            return decision == ApprovalDecision.APPROVED
        except asyncio.TimeoutError:
            logger.warning(f"Approval timeout for request {request_id}")
            return False
        finally:
            self._pending.pop(request_id, None)
    
    async def resolve(self, request_id: str, decision: ApprovalDecision):
        """Called by the approval UI/webhook when a human makes a decision."""
        if request_id in self._pending:
            self._pending[request_id].set_result(decision)

# mcp_sandbox.py
import subprocess
import os
from pathlib import Path

class SandboxedMCPServer:
    """Runs MCP servers in Docker containers with strict resource limits."""
    
    def __init__(self, server_command: str, container_name: str,
                 memory_limit: str = "256m", cpu_limit: float = 0.5,
                 network_access: bool = False):
        self.server_command = server_command
        self.container_name = container_name
        self.memory_limit = memory_limit
        self.cpu_limit = cpu_limit
        self.network_access = network_access
        self._process: subprocess.Popen | None = None
    
    def start(self) -> tuple[int, int]:
        """Start the sandboxed server and return (stdin_fd, stdout_fd)."""
        docker_args = [
            "docker", "run", "--rm",
            f"--name={self.container_name}",
            f"--memory={self.memory_limit}",
            f"--cpus={self.cpu_limit}",
            "--read-only",
            "--cap-drop=ALL",
            "--security-opt=no-new-privileges",
        ]
        
        if not self.network_access:
            docker_args.append("--network=none")
        
        # Mount necessary files read-only
        docker_args.extend([
            "-v", f"{self._get_server_path()}:/server:ro",
            "--workdir", "/server",
            "python", "-m", "mcp_server"
        ])
        
        self._process = subprocess.Popen(
            docker_args,
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            bufsize=0
        )
        
        return self._process.stdin.fileno(), self._process.stdout.fileno()
    
    def stop(self):
        """Gracefully stop the sandboxed server."""
        if self._process:
            self._process.terminate()
            try:
                self._process.wait(timeout=10)
            except subprocess.TimeoutExpired:
                self._process.kill()

# mcp_pool.py
import asyncio
from collections import deque
from contextlib import asynccontextmanager

class MCPSessionPool:
    """
    Connection pool for MCP servers that maintains session isolation.
    Critical for multi-tenant deployments where each tenant needs their own context.
    """
    
    def __init__(self, server_params: StdioServerParameters, 
                 pool_size: int = 20, max_queue_size: int = 100):
        self.server_params = server_params
        self.pool_size = pool_size
        self.max_queue_size = max_queue_size
        self._pool: asyncio.Queue[ClientSession] = asyncio.Queue(maxsize=pool_size)
        self._initialized = False
        self._lock = asyncio.Lock()
    
    async def initialize(self):
        """Pre-warm the pool with initial connections."""
        async with self._lock:
            if self._initialized:
                return
            tasks = [self._create_session() for _ in range(self.pool_size)]
            results = await asyncio.gather(*tasks, return_exceptions=True)
            for result in results:
                if not isinstance(result, Exception):
                    await self._pool.put(result)
            self._initialized = True
    
    async def _create_session(self) -> ClientSession:
        """Create a new MCP client session."""
        from mcp import stdio_client
        async with stdio_client(self.server_params) as (read, write):
            session = ClientSession(read, write)
            await session.initialize()
            return session
    
    @asynccontextmanager
    async def get_session(self) -> AsyncGenerator[ClientSession, None]:
        """
        Borrow a session from the pool.
        Sessions are returned to the pool after use for reuse.
        """
        try:
            session = await asyncio.wait_for(self._pool.get(), timeout=10.0)
        except asyncio.TimeoutError:
            # Pool exhausted - create an overflow session
            session = await self._create_session()
            overflow = True
        else:
            overflow = False
        
        try:
            yield session
        except Exception:
            # Session may be corrupted - replace it
            if not overflow:
                await self._replace_session(session)
            raise
        else:
            if not overflow:
                await self._pool.put(session)
    
    async def _replace_session(self, old_session: ClientSession):
        """Replace a corrupted session with a fresh one."""
        try:
            await old_session.close()
        except Exception:
            pass
        new_session = await self._create_session()
        await self._pool.put(new_session)
    
    async def shutdown(self):
        """Gracefully shut down all pooled sessions."""
        while not self._pool.empty():
            session = self._pool.get_nowait()
            try:
                await session.close()
            except Exception:
                pass

# mcp_router.py
from typing import Protocol

class MCPToolProvider(Protocol):
    """Interface for any MCP server that provides tools."""
    
    async def list_tools(self) -> list[dict]:
        ...
    
    async def call_tool(self, tool_name: str, arguments: dict) -> str:
        ...

class MCPRouter:
    """Routes tool calls to the appropriate MCP server."""
    
    def __init__(self):
        self._providers: dict[str, MCPToolProvider] = {}
        self._tool_index: dict[str, str] = {}  # tool_name -> provider_name
    
    def register_provider(self, name: str, provider: MCPToolProvider):
        self._providers[name] = provider
    
    async def build_tool_index(self):
        """Build an index of all available tools across all providers."""
        for name, provider in self._providers.items():
            tools = await provider.list_tools()
            for tool in tools:
                tool_name = tool["name"]
                # Handle naming conflicts with prefixing
                if tool_name in self._tool_index:
                    prefixed_name = f"{name}__{tool_name}"
                    self._tool_index[prefixed_name] = name
                    tool["name"] = prefixed_name
                else:
                    self._tool_index[tool_name] = name
    
    async def route_call(self, tool_name: str, arguments: dict) -> str:
        """Route a tool call to the correct provider."""
        provider_name = self._tool_index.get(tool_name)
        if not provider_name:
            raise UnknownToolError(f"Tool '{tool_name}' not found")
        
        provider = self._providers[provider_name]
        return await provider.call_tool(tool_name, arguments)
    
    def get_all_tools(self) -> list[dict]:
        """Return all tools with their provider information."""
        all_tools = []
        for tool_name, provider_name in self._tool_index.items():
            provider = self._providers[provider_name]
            # Fetch full tool definition
            tools = provider.list_tools()
            for tool in tools:
                if tool["name"] == tool_name:
                    tool["provider"] = provider_name
                    all_tools.append(tool)
        return all_tools

# mcp_state.py
from dataclasses import dataclass, field
from typing import Any

@dataclass
class AgentCheckpoint:
    session_id: str
    step: int
    completed_actions: list[dict] = field(default_factory=list)
    pending_actions: list[dict] = field(default_factory=list)
    context: dict[str, Any] = field(default_factory=dict)
    tool_results_cache: dict[str, str] = field(default_factory=dict)
    
    def to_dict(self) -> dict:
        return {
            "session_id": self.session_id,
            "step": self.step,
            "completed_actions": self.completed_actions,
            "pending_actions": self.pending_actions,
            "context": self.context,
            "tool_results_cache": self.tool_results_cache
        }
    
    @classmethod
    def from_dict(cls, data: dict) -> "AgentCheckpoint":
        return cls(**data)

class CheckpointStore:
    """Persists agent state for crash recovery and session resumption."""
    
    def __init__(self, backend: str = "redis"):
        self._backend = backend
        self._prefix = "mcp:agent:checkpoint:"
    
    async def save(self, checkpoint: AgentCheckpoint):
        """Save a checkpoint to persistent storage."""
        data = checkpoint.to_dict()
        if self._backend == "redis":
            await self._redis.set(
                self._prefix + checkpoint.session_id,
                json.dumps(data),
                ex=86400  # 24 hour TTL
            )
    
    async def load(self, session_id: str) -> AgentCheckpoint | None:
        """Load the most recent checkpoint for a session."""
        if self._backend == "redis":
            data = await self._redis.get(self._prefix + session_id)
            if data:
                return AgentCheckpoint.from_dict(json.loads(data))
        return None
    
    async def resume_execution(self, session_id: str, 
                               agent: "MCPAgent") -> None:
        """Resume agent execution from the last checkpoint."""
        checkpoint = await self.load(session_id)
        if not checkpoint:
            raise NoCheckpointError(f"No checkpoint found for {session_id}")
        
        # Restore tool results cache to avoid redundant API calls
        agent.tool_results_cache = checkpoint.tool_results_cache
        agent.completed_actions = checkpoint.completed_actions
        
        # Continue from the next pending action
        for action in checkpoint.pending_actions:
            result = await agent.execute_action(action)
            await self.save(agent.current_checkpoint())

# production_agent.py
import asyncio
import json
from dataclasses import dataclass
from typing import Any, Callable

@dataclass
class AgentConfig:
    model: str = "gpt-4-turbo"
    max_steps: int = 20
    temperature: float = 0.0
    tool_timeout: float = 30.0
    enable_checkpointing: bool = True
    require_approval_above: str = "medium"  # risk level threshold

class ProductionMCPAgent:
    """
    A production-ready MCP agent with lifecycle management,
    observability, security, and state persistence.
    """
    
    def __init__(self, config: AgentConfig, 
                 mcp_params: StdioServerParameters,
                 approval_gate: ApprovalGate | None = None):
        self.config = config
        self.mcp_params = mcp_params
        self.approval_gate = approval_gate
        
        # Infrastructure components
        self.pool = MCPSessionPool(mcp_params, pool_size=5)
        self.tracer = MCPTracer(service_name="production-agent")
        self.checkpoint_store = CheckpointStore()
        self.policy = ToolPermissionPolicy()
        
        # Agent state
        self.session_id: str | None = None
        self.step: int = 0
        self.tool_results_cache: dict[str, str] = {}
        self.completed_actions: list[dict] = []
    
    async def run(self, user_input: str, session_id: str | None = None) -> str:
        """Execute the agent loop with full production safeguards."""
        self.session_id = session_id or str(uuid4())
        
        # Attempt to resume from checkpoint
        if self.config.enable_checkpointing and session_id:
            checkpoint = await self.checkpoint_store.load(session_id)
            if checkpoint:
                self.step = checkpoint.step
                self.tool_results_cache = checkpoint.tool_results_cache
                self.completed_actions = checkpoint.completed_actions
        
        await self.pool.initialize()
        
        messages = [{"role": "user", "content": user_input}]
        final_answer = None
        
        while self.step < self.config.max_steps:
            self.step += 1
            
            # Get model response
            response = await self._call_model(messages)
            
            if response.tool_calls:
                for tool_call in response.tool_calls:
                    result = await self._execute_tool_call(
                        tool_call, 
                        self.session_id
                    )
                    messages.append(result)
            else:
                final_answer = response.content
                break
            
            # Checkpoint after each step
            if self.config.enable_checkpointing:
                await self._save_checkpoint()
        
        return final_answer or "Max steps reached without final answer"
    
    async def _execute_tool_call(self, tool_call, session_id: str) -> dict:
        """Execute a tool call with security checks and observability."""
        tool_name = tool_call.function.name
        arguments = json.loads(tool_call.function.arguments)
        
        # Security: validate against policy
        validated_args = self.policy.validate_call(tool_name, arguments)
        
        # Security: check approval gate
        risk_level = self.policy.get_risk_level(tool_name)
        if self.approval_gate and risk_level in ("high", "critical"):
            approved = await self.approval_gate.check(
                tool_name=tool_name,
                arguments=validated_args,
                risk_level=risk_level,
                description=f"Agent requests {tool_name}",
                agent_id=self.session_id,
                session_id=session_id
            )
            if not approved:
                return {
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": "Action denied by approval gate"
                }
        
        # Cache check
        cache_key = f"{tool_name}:{json.dumps(validated_args, sort_keys=True)}"
        if cache_key in self.tool_results_cache:
            return {
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": self.tool_results_cache[cache_key]
            }
        
        # Execute with tracing
        async with self.pool.get_session() as session:
            result = await self.tracer.traced_call(
                session, tool_name, validated_args, session_id
            )
        
        result_text = result.content[0].text
        self.tool_results_cache[cache_key] = result_text
        self.completed_actions.append({
            "tool": tool_name,
            "arguments": validated_args,
            "result_length": len(result_text)
        })
        
        return {
            "role": "tool",
            "tool_call_id": tool_call.id,
            "content": result_text
        }
    
    async def _call_model(self, messages: list[dict]) -> Any:
        """Call the LLM with current message history."""
        # Implementation depends on your LLM provider
        ...
    
    async def _save_checkpoint(self):
        """Persist current state for crash recovery."""
        checkpoint = AgentCheckpoint(
            session_id=self.session_id,
            step=self.step,
            completed_actions=self.completed_actions,
            tool_results_cache=self.tool_results_cache
        )
        await self.checkpoint_store.save(checkpoint)

# k8s-mcp-agent.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: mcp-agent
spec:
  replicas: 3
  selector:
    matchLabels:
      app: mcp-agent
  template:
    metadata:
      labels:
        app: mcp-agent
    spec:
      containers:
      - name: agent
        image: your-registry/mcp-agent:latest
        resources:
          requests:
            cpu: "500m"
            memory: "512Mi"
          limits:
            cpu: "2"
            memory: "1Gi"
        env:
        - name: MCP_SERVER_COMMAND
          value: "python -m mcp_server"
        - name: POOL_SIZE
          value: "10"
        - name: REDIS_URL
          valueFrom:
            secretKeyRef:
              name: redis-credentials
              key: url
        livenessProbe:
          httpGet:
            path: /health
            port: 8080
          initialDelaySeconds: 30
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /ready
            port: 8080
          initialDelaySeconds: 15
          periodSeconds: 5
---
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: mcp-agent-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: mcp-agent
  minReplicas: 3
  maxReplicas: 20
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70
  - type: Pods
    pods:
      metric:
        name: mcp_active_sessions
      target:
        type: AverageValue
        averageValue: "50"

# Loop detection
class LoopDetector:
    def __init__(self, max_repeats: int = 3):
        self.max_repeats = max_repeats
        self._action_history: list[str] = []
    
    def check(self, action: dict) -> bool:
        """Returns True if this action appears to be looping."""
        action_key = json.dumps(action, sort_keys=True)
        self._action_history.append(action_key)
        
        # Check for repeated sequences
        recent = self._action_history[-self.max_repeats * 2:]
        for i in range(len(recent) - self.max_repeats):
            if (recent[i:i+self.max_repeats] == 
                recent[i+self.max_repeats:i+self.max_repeats*2]):
                return True
        return False

def validate_tool_arguments(tool_name: str, arguments: dict, 
                            tool_schema: dict) -> dict:
    """Validate arguments against the tool's JSON schema."""
    errors = []
    
    # Check for unknown arguments
    known_args = tool_schema.get("properties", {})
    for key in arguments:
        if key not in known_args:
            errors.append(f"Unknown argument: '{key}'")
    
    # Check for missing