Agent Runtime¶

agent_runtime — mvp.agent_runtime

Cluster: Uncategorised | Type: component | MCP Tools: 14

Overview¶

Shared production infrastructure for all G6 agent blocks. Provides lifecycle management, retry with exponential backoff, circuit breaker recovery, backend fallback routing, structured logging with correlation IDs, metrics collection, and an HTTP /metrics endpoint for Prometheus scraping.

When to use:

• Wrapping agent blocks with production-grade resilience: retry, circuit breaker recovery, and fallback.
• Collecting per-agent latency, token, circuit state, and error metrics.
• Correlating logs across a multi-step pipeline using correlation IDs.

Example:

from mvp.agent_runtime import CircuitBreaker, RetryPolicy, FallbackChain, BackendEntry

# Retry with exponential backoff
policy = RetryPolicy(max_attempts=3, initial_delay=1.0, max_delay=30.0)

# Circuit breaker - opens after 5 failures and probes recovery after 60s
breaker = CircuitBreaker(failure_threshold=5, cooldown_seconds=60.0)

# Fallback chain - tries backends in order
chain = FallbackChain(backends=[
    BackendEntry(name="ollama", fn=ollama_call, circuit=breaker),
    BackendEntry(name="openrouter", fn=openrouter_call),
])

result, backend_name = chain.run(prompt)

Works well with: llm_router, observability, telemetry

Public API¶

AgentRuntimeInput(BaseModel)¶

AgentRuntimeOutput(BaseModel)¶

Methods:

model_post_init(__context: Any) -> None¶

AgentRuntimeBlock(AIBlock)¶

AIBlock wrapper exposing agent runtime status, health, and metrics.

Methods:

infer(input: AgentRuntimeInput) -> Result[AgentRuntimeOutput]¶

CircuitState(Enum)¶

CircuitBreaker¶

Methods:

state() -> CircuitState¶

Current circuit state. Does not claim the HALF_OPEN probe slot.

acquire_probe() -> bool¶

Atomically claim the HALF_OPEN probe slot.

record_success() -> None¶

Record a successful call and reset to CLOSED.

record_failure() -> None¶

record_transient() -> None¶

Record a transient provider error (HTTP 408/429/500/502/503/504,

record_transient_or_failure(error: object) -> None¶

Route error to :meth:record_transient when it is a transient

is_available() -> bool¶

Return True when a request may run, including one HALF_OPEN probe.

BackendEntry¶

FallbackChain¶

Methods:

run(*args: Any, **kwargs: Any) -> tuple[Any, str]¶

Try backends in order. Returns (result, backend_name).

LifecycleMixin¶

Non-dataclass mixin providing lifecycle management to agent blocks.

Methods:

start() -> Result[None]¶

Start the block: register logger, start metrics server, call _start_hook().

stop() -> Result[None]¶

Stop the block: flush, decrement metrics server ref count.

health_check() -> Result[dict[str, Any]]¶

Return health status dict. Safe to call at any time.

save(path: str) -> Result[str]¶

Serialize circuit state + resource_bounds config to JSON at path.

load(path: str) -> Result[bool]¶

Restore state from path (best-effort — does not restore circuit failures).

ingest(data: Any) -> Result[Any]¶

Agent blocks do not use stream I/O. Returns fail.

emit() -> Result[Any]¶

Agent blocks do not use stream I/O. Returns fail.

AgentLogger¶

Structured JSON logger bound to a specific agent name.

Constructor:

Methods:

info(event: str, **kwargs: object) -> None¶

warning(event: str, **kwargs: object) -> None¶

error(event: str, **kwargs: object) -> None¶

debug(event: str, **kwargs: object) -> None¶

MetricsCollector¶

Methods:

record_infer(agent: str, status: str, latency_ms: float, tokens_in: int = 0, tokens_out: int = 0) -> None¶

record_circuit(agent: str, backend: str, state: int) -> None¶

state: 0=CLOSED, 1=OPEN, 2=HALF_OPEN

prometheus_text() -> str¶

snapshot() -> dict¶

Structured, JSON-serialisable view of the same metrics that

RetryPolicy¶

Retry a callable with exponential backoff and optional jitter.

Methods:

run(fn: Callable[..., Any], *args: Any, **kwargs: Any) -> Any¶

Call fn(args, *kwargs) with retry on retryable_exceptions.

Functions¶

set_correlation_id(cid: str) -> None¶

Set the correlation ID for the current execution context.

get_correlation_id() -> str¶

Get the correlation ID for the current execution context.

get_collector() -> MetricsCollector¶

Return the module-level singleton MetricsCollector.

start_metrics_server(port: int = 9090, host: str = '127.0.0.1') -> None¶

Start the metrics server (or increment reference count if already running).

stop_metrics_server() -> None¶

Decrement reference count. Shutdown server when count reaches 0.

MCP Tools¶