Multi-Agent Shared State
Thread-safe SharedBudget and SharedCircuitBreaker for multi-agent systems.
Overview
When multiple agents share tools, you often need shared guardrail state — a single budget pool or circuit breaker that applies across all agents. The agentguard.shared module provides thread-safe SharedBudget and SharedCircuitBreaker objects.
SharedBudget
Both tools draw from the same budget pool:
python
from agentguard.shared import SharedBudget
from agentguard import guard
# Create a shared budget pool
budget = SharedBudget(
max_cost_per_session=10.00,
max_calls_per_session=100,
)
# Both agents draw from the same pool
@guard(budget=budget.config)
def agent1_search(q: str) -> dict:
return requests.get(f"https://api.search.com?q={{q}}").json()
@guard(budget=budget.config)
def agent2_search(q: str) -> dict:
return requests.get(f"https://api.search.com?q={{q}}").json()
# After some calls:
print(budget.session_spend) # Combined spend from both tools
print(budget.session_calls) # Combined call count
print(budget.stats()) # Full SharedBudgetStats snapshotSharedBudget API
| Method / Property | Description |
|---|---|
.config | BudgetConfig to pass to @guard(budget=...) |
.session_spend | Total USD spent across all tools |
.session_calls | Total call count |
.stats() | Returns SharedBudgetStats snapshot |
.reset() | Reset spend and calls to zero |
.check_pre_call() | Check if next call would exceed limits |
SharedCircuitBreaker
A single circuit breaker shared across multiple tools:
python
from agentguard.shared import SharedCircuitBreaker
from agentguard import guard
cb = SharedCircuitBreaker(
failure_threshold=10,
recovery_timeout=30,
)
@guard(circuit_breaker=cb.config)
def agent1_call(x: str) -> dict:
return external_api.call(x)
@guard(circuit_breaker=cb.config)
def agent2_call(x: str) -> dict:
return external_api.call(x)
# If agent1 causes 10 failures, agent2's circuit opens too
print(cb.circuit_state) # CircuitState.OPEN
print(cb.stats()) # SharedCircuitStats with full state infoThread-Safety Guarantees
Both SharedBudget and SharedCircuitBreaker are fully thread-safe:
- All state updates are protected by
threading.Lock - Global registries use a module-level lock
- Instances register themselves in a global registry keyed by
shared_id - Use
clear_shared_registry()in tests to ensure a clean slate
python
from agentguard.shared import get_shared_budget, get_shared_circuit_breaker, clear_shared_registry
# Look up a shared instance by ID
budget = get_shared_budget("budget-abc123")
cb = get_shared_circuit_breaker("cb-def456")
# Clean up in tests
clear_shared_registry()