Agent Security & MCP Blueprints

Model Context Protocol (MCP) enables LLMs to query external databases, run code, and invoke APIs. However, because agents act autonomously on untrusted inputs (such as emails, scraped web pages, or customer chats), MCP servers are primary targets for **indirect prompt injection** and **privilege escalation**.

Core Defenses for MCP Implementations:

• Transport Layer Isolation: Run local MCP servers via standard input/output (`stdio`) and restrict network ports. If using Server-Sent Events (SSE), enforce strict token authentication.
• Read-Only Scopes by Default: Never grant write or execute access on file system or database tools unless explicitly requested and backed by human-in-the-loop validation.
• Sanitize Prompt Ingress: Scan all incoming variables sent to MCP servers. Payload payloads carrying command sequences (e.g., `rm -rf`, `DROP TABLE`, or Instruction Overrides) must be filtered.

Recommended Architecture:

When an external API or database tool fails or times out, an AI agent will often enter a runaway recursive loop, repeatedly invoking the same broken tool and draining your API budget. A **circuit breaker** pattern blocks calls to failing tools immediately, preventing loop propagation.

Reference Circuit Breaker Blueprints:

class ToolCircuitBreaker {
  private failures = 0;
  private state: 'CLOSED' | 'OPEN' | 'HALF_OPEN' = 'CLOSED';
  private lastStateChange = Date.now();
  private readonly threshold = 3; // Max failures
  private readonly cooldown = 8000; // 8 seconds

  async execute<T>(toolCall: () => Promise<T>): Promise<T> {
    if (this.state === 'OPEN') {
      if (Date.now() - this.lastStateChange > this.cooldown) {
        this.state = 'HALF_OPEN';
      } else {
        throw new Error('Circuit breaker is OPEN. Blocked tool execution.');
      }
    }

    try {
      const result = await toolCall();
      this.reset();
      return result;
    } catch (error) {
      this.handleFailure();
      throw error;
    }
  }

  private handleFailure() {
    this.failures++;
    if (this.failures >= this.threshold) {
      this.state = 'OPEN';
      this.lastStateChange = Date.now();
    }
  }

  private reset() {
    this.failures = 0;
    this.state = 'CLOSED';
  }
}

import time

class CircuitBreaker:
    def __init__(self, threshold=3, cooldown=8):
        self.threshold = threshold
        self.cooldown = cooldown
        self.failures = 0
        self.state = "CLOSED"  # CLOSED, OPEN, HALF_OPEN
        self.last_state_change = time.time()

    def execute(self, tool_func, *args, **kwargs):
        if self.state == "OPEN":
            if time.time() - self.last_state_change > self.cooldown:
                self.state = "HALF_OPEN"
            else:
                raise Exception("Circuit breaker is OPEN. Blocked execution.")

        try:
            result = tool_func(*args, **kwargs)
            self.reset()
            return result
        except Exception as e:
            self.handle_failure()
            raise e

    def handle_failure(self):
        self.failures += 1
        if self.failures >= self.threshold:
            self.state = "OPEN"
            self.last_state_change = time.time()

    def reset(self):
        self.failures = 0
        self.state = "CLOSED"

*Note: These mechanisms are built natively into **Outpost Tier 0** to protect agent operations automatically without requiring custom wrapper boilerplate code in your application layer.*