agent-prod

English | 简体中文 | Design | MCP Integration | Usage

Quality gates for production AI agents. agent-prod wraps any agent run or release in an 8-gate safety pipeline — permission, budget, trace integrity, regression, gray release, audit, answer quality, and execution consistency. Like tests for code, but for agent behavior.

from agent_prod import trace

result = trace(
    agent="my-agent",
    session_id="session-001",
    current_metrics={"final_response": "Paris", "success_rate": 0.99},
)
print(result["status"])  # "production" | "rejected"

Pipeline

Agent run ──▶ Gate0 ──▶ Gate1 ──▶ Gate2 ──▶ Gate3 ──▶ Gate4 ──▶ Gate5 ──▶ Gate6 ──▶ Gate7 ──▶ Approve?
              │         │          │          │          │          │          │          │
             risk     budget    trace     regress.   gray      audit     answer    exec.
             ACL      check     DAG       compare   rollout   policy    quality   consistency

How this differs from eval frameworks: Eval frameworks score a single dimension (answer correctness) in isolation. agent-prod closes the loop — permission → budget → trace → regression → release → audit → quality → consistency — and rejects the run at the first failure. This is the difference between "this answer scores 0.85" and "this agent run is not safe for production." See design philosophy →

Quick Start

# Install
pip install agent-prod

# Evaluate one trace — no server needed
python -c "
from agent_prod import trace
result = trace(
    agent='demo',
    session_id='demo-1',
    decisions=[{
        'decision_id': 'd1',
        'model': 'gpt-4',
        'prompt_tokens': 100,
        'completion_tokens': 50,
        'tool_calls': [
            {'tool_name': 'web_search', 'arguments': {'q': 'weather'}, 'success': True},
        ],
    }],
    current_metrics={'final_response': 'Sunny, 22°C', 'success_rate': 1.0},
    human_approver='demo',
)
print(f'Passed: {result[\"status\"]}')
"

# Or start the server
agent-prod configure
agent-prod serve
python examples/basic_trace.py
agent-prod stats

MCP Server — Use From Any Agent

agent-prod exposes all 8 gates as MCP tools. Claude Desktop, Cursor, Cline — any MCP client can call quality-gate evaluations directly.

pip install "agent-prod[mcp]"
agent-prod-mcp

// claude_desktop_config.json
{ "mcpServers": { "agent-prod": { "command": "agent-prod-mcp" } } }

→ Full MCP integration guide

MCP Registry — Publish, Search, Install

# Publish your MCP server
agent-prod registry publish my-server \
    --command "uvx my-server" \
    --description "Search and index documentation" \
    --tags "search,docs"

# Search for servers
agent-prod registry search mcp

# List local registry
agent-prod registry list

→ Registry source

Agent Observability — OpenTelemetry Spans

Wrap pipeline evaluations with OpenTelemetry spans. Each gate becomes a span under an agent-run trace, exportable to any OTLP-compatible backend (Grafana, Honeycomb, SigNoz).

from agent_prod.observability.otel import AgentSpanExporter

exporter = AgentSpanExporter(endpoint="http://localhost:4317")
exporter.export_pipeline(improvement, agent_type="hermes")
# → Gate0–Gate7 spans exported to your observability backend

Zero hard dependency — works without opentelemetry installed. Activate with pip install opentelemetry-api opentelemetry-sdk opentelemetry-exporter-otlp.

A2A — Agent-to-Agent Delegation

Delegate tasks between agents via a lightweight protocol with capability negotiation, partial-success semantics, and error chain attribution.

from agent_prod.a2a import A2AAgent, A2ATask, A2ADelegator

class SearchAgent(A2AAgent):
    capabilities = ["web_search", "news"]

    def execute(self, task: A2ATask):
        results = search(task.input["q"])
        return {"results": results}

delegator = A2ADelegator()
delegator.register(SearchAgent())

task = A2ATask(name="search", input={"q": "weather"}, required_capabilities=["web_search"])
result = delegator.delegate(task)

Comes with a LangChain adapter (create_langchain_tool) to plug into existing agent pipelines.

Multi-Agent Trust Chain — Phase 5

When a parent agent delegates to a child, the child inherits a subset of the parent's authority — never more. TrustChainValidator enforces this at Gate0:

from agent_prod.gates.trust_chain import TrustChainValidator, TaskACL, TrustLevel

tc = TrustChainValidator()
tc.register_task(TaskACL(
    task_id="task-001",
    parent_agent="qclaw",
    child_agent="code-reviewer",
    trust_level=TrustLevel.RESTRICTED,
    allowed_tools={"Read", "Grep"},          # child can only call these
    allowed_domains={"finance"},             # child can only operate in this domain
    expires_at=datetime.now(UTC) + timedelta(hours=1),
))

# Gate0 now blocks any tool call from `child_agent=code-reviewer`
# outside the ACL — even if it's declared in declared_tools.

Without a registered ACL, the validator stays out of the way — Gate0 falls back to its standard declared-tools + auth-grant checks. So trust-chain enforcement is opt-in per delegation relationship.

Async Approval Workflow — Phase 3

Gate5 can emit a pending_approval state instead of binary pass/fail. The pipeline halts, the improvement is persisted, and the pipeline resumes on external decision — via HTTP callback, CLI, or webhook.

# Pipeline reaches Gate5 with only "Human approval" missing
# → status: pending_approval, improvement persisted

# Approve via HTTP
curl -X POST http://localhost:8080/v1/approvals/$APPROVAL_ID/decide \
  -H "Content-Type: application/json" \
  -d '{"approved": true, "approver": "alice", "reason": "looks good"}'
# → runs Gate6, Gate7, promotes to PRODUCTION

# Or reject
curl -X POST http://localhost:8080/v1/approvals/$APPROVAL_ID/decide \
  -d '{"approved": false, "approver": "bob", "reason": "rollback plan missing"}'
# → status: rejected, fail_gate: gate5_approval

Approval records are idempotent — deciding twice raises ValueError. Records auto-expire after a configurable TTL (default 24h).

CLI

agent-prod approval list                       # list pending approvals
agent-prod approval list --agent qclaw         # filter by agent
agent-prod approval list --status pending      # filter by status
agent-prod approval approve <id> --approver alice --reason "LGTM"
agent-prod approval reject <id> --approver bob

Server-restart rehydration

The in-memory ApprovalQueue is lost on server restart, but Improvement records with status=PENDING_APPROVAL persist in the repository. On startup, the server scans the repository for pending approvals and re-registers them in the queue — so callbacks received after a restart still resume the pipeline.

Domain Policy Engine — Phase 2

Industry-specific risk escalation at Gate0. The base tool risk is only escalated, never downgraded when an agent operates in a regulated domain.

# config.yaml
gates:
  domain_policy:
    enabled: true
    domains:
      finance:
        risk_overrides:
          web_search: elevated   # benign → elevated in finance context
        required_compliance: [sox, pci_dss]
      medical:
        risk_overrides:
          read_file: elevated
        required_compliance: [hipaa]

from agent_prod.gates.domain_policy import DomainPolicyEngine

engine = DomainPolicyEngine(config)
result = engine.get_effective_risk("web_search", "my-agent", domain="finance")
# RiskLevel.ELEVATED  (was BENIGN in vanilla Gate0)

# Compliance claims are enforced: missing HIPAA claim in medical domain → block
cr = engine.validate_compliance_claims(
    tool="read_file", domain="medical",
    compliance_claims={},  # missing HIPAA
    agent_type="my-agent",
)
assert cr.violation  # ViolationType.MISSING_COMPLIANCE

Observability — Prometheus + Grafana — Phase 4

# Spin up the full monitoring stack
docker compose --profile observability up -d
# → Prometheus on :9090, Grafana on :3001

Pre-built dashboard (deploy/grafana/dashboards/agent_prod_overview.json) auto-provisions with panels for: pipeline outcomes, per-gate rejection rates, gate1 circuit breaker state, domain-policy escalations, and approval queue depth. The dashboard is provisioned automatically on first Grafana boot — no manual import required.

Key Prometheus metrics exposed on :8080/metrics:

GatePlugin Interface — Extend With One Class

Every gate is a plug-in. Write your own gate in ~30 lines:

from agent_prod.gates.interface import GatePlugin, register_gate
from agent_prod.gates.models import GateName, GateResult, Improvement

class MyCustomGate(GatePlugin):
    name = GateName("my_custom_gate")
    rollback_level = RollbackLevel.L1

    def verify(self, improvement: Improvement) -> GateResult:
        if improvement.candidate_output.get("my_field", 0) >= 90:
            return GateResult(gate_name=self.name, passed=True, reason="OK")
        return GateResult(gate_name=self.name, passed=False, reason="my_field < 90")

    def rollback(self, improvement: Improvement) -> None:
        pass

    @classmethod
    def from_config(cls, config, name):
        return cls()

register_gate(GateName("my_custom_gate"), MyCustomGate)

The engine discovers gates through the GatePlugin ABC — no monkey-patching, no framework fork. Add a gate, register it, and from_yaml() picks it up. → Full interface design →

Proof Points

Why Not Just an Eval Framework?

Eval frameworks answer "how good is this output". agent-prod answers "is this agent run safe for production".

Deployment

# One command — Postgres + agent-prod + MCP
docker compose up -d

See docker-compose.yml and .env.example.

Start Here

• Design document — architecture decisions, GatePlugin ABC, and pipeline topology
• MCP Integration guide — Claude Desktop, Cursor, Cline, Hermes setup
• MCP Registry — publish, search, and install MCP servers
• A2A Protocol — agent-to-agent delegation
• Observability — OpenTelemetry spans for agent runs
• Examples — runnable traces and release scenarios
• Usage guide — CLI, configuration, Gate0–Gate7 details
• Dogfood report — we ate our own dog food
• Calibration guide — tuning Gate5/Gate6 for your agents
• Roadmap — production validation plan and next proof points

License

MIT License. See LICENSE.