API reference

Core

AgentGuard — inter-agent security firewall for multi-agent AI systems.

class agentguard.AgentGuard(risk_threshold=0.85, enable_trust_attestation=True, enable_capability_enforcement=True, audit_log_path='./audit.jsonl', mode='enforce', task_objective=None, enable_consistency_check=True, consistency_threshold=0.1, consistency_ml_risk_floor=0.75, model_path=None, require_ml_model=False, enable_otel_export=False, include_message_preview=True, max_message_bytes=1048576)[source]

Bases: object

Inter-agent security middleware entry point.

Parameters:

  • risk_threshold (float)

  • enable_trust_attestation (bool)

  • enable_capability_enforcement (bool)

  • audit_log_path (str | None)

  • mode (str)

  • task_objective (str | None)

  • enable_consistency_check (bool)

  • consistency_threshold (float)

  • consistency_ml_risk_floor (float)

  • model_path (str | None)

  • require_ml_model (bool)

  • enable_otel_export (bool)

  • include_message_preview (bool)

  • max_message_bytes (int)

check_data_source(agent_id, data_source)[source]

Validate data-source access; False when blocked in enforce mode.

Parameters:

  • agent_id (str)

  • data_source (str)

Return type:

bool

check_endpoint(agent_id, endpoint)[source]

Validate an external endpoint; False when blocked in enforce mode.

Parameters:

  • agent_id (str)

  • endpoint (str)

Return type:

bool

check_output_tokens(agent_id, token_count)[source]

Validate output size; False when blocked in enforce mode.

Parameters:

  • agent_id (str)

  • token_count (int)

Return type:

bool

check_tool_call(agent_id, tool_name, *, endpoint=None)[source]

Validate a tool call; False when blocked in enforce mode.

Parameters:

  • agent_id (str)

  • tool_name (str)

  • endpoint (str | None)

Return type:

bool

get_manifest(agent_id)[source]

Return the registered capability manifest for an agent, if any.

Parameters:

agent_id (str)

Return type:

CapabilityManifest | None

inspect_content(sender_id, recipient_id, message)[source]

Inspect unsigned content (rules + ML + consistency, no trust check).

Use this at boundaries where messages cannot carry Ed25519 signatures — user input, framework middleware hooks, or third-party content. The decision is audited like any other inspection.

Parameters:

  • sender_id (str)

  • recipient_id (str)

  • message (str)

Return type:

FirewallDecision

inspect_message(sender_id, recipient_id, message, payload_bytes, *, signature=None)[source]

Run inspection, trust, and size checks on a message.

Parameters:

  • sender_id (str)

  • recipient_id (str)

  • message (str)

  • payload_bytes (bytes)

  • signature (bytes | None)

Return type:

FirewallDecision

inspect_tool_output(tool_name, output, agent_id)[source]

Inspect a tool return value for poisoning; audits the decision.

Parameters:

  • tool_name (str)

  • output (str)

  • agent_id (str)

Return type:

FirewallDecision

property is_ml_model_loaded: bool

Return True when the DeBERTa ONNX scorer is active.

register_agent(agent_id, manifest)[source]

Register agent identity, trust keys, and capability manifest.

Parameters:
Return type:

None

register_delegated_agent(parent_agent_id, child_manifest)[source]

Register a sub-agent with monotonic capability attenuation.

Parameters:
Return type:

CapabilityManifest

rotate_keys()[source]

Rotate ephemeral trust keys for all registered agents.

Return type:

None

sign_payload(sender_id, payload_bytes, *, recipient_id, nonce=None)[source]

Sign an inter-agent envelope (sender, recipient, nonce, payload).

The returned wire signature binds the recipient and a unique nonce so a previously signed payload cannot be redirected or silently replayed. Pass the bytes unchanged to inspect_message() as signature.

Parameters:
Return type:

bytes

wrap(graph)[source]

Wrap a LangGraph compiled graph with AgentGuard interception.

Parameters:

graph (object)

Return type:

object

wrap_mcp_tool(tool_fn, agent_id)[source]

Wrap a sync MCP tool callable with output inspection.

Parameters:

  • tool_fn (F)

  • agent_id (str)

Return type:

F

wrap_mcp_tool_async(tool_fn, agent_id)[source]

Wrap an async MCP tool callable with output inspection.

Parameters:

  • tool_fn (AF)

  • agent_id (str)

Return type:

AF

exception agentguard.AgentGuardException(action, sender_id, recipient_id, risk_score, trust_result, capability_result, failure_reason)[source]

Bases: Exception

Raised when AgentGuard blocks or quarantines an inter-agent message.

Parameters:

  • action (str)

  • sender_id (str)

  • recipient_id (str)

  • risk_score (float)

  • trust_result (str)

  • capability_result (str)

  • failure_reason (str | None)

Return type:

None

action: str
capability_result: str
failure_reason: str | None
recipient_id: str
risk_score: float
sender_id: str
trust_result: str
class agentguard.CapabilityManifest(agent_id, permitted_tools=<factory>, forbidden_tools=<factory>, allowed_data_sources=<factory>, permitted_endpoints=<factory>, max_output_tokens=4096, external_contact=False, can_spawn_agents=False, max_delegation_depth=0)[source]

Bases: object

Declarative runtime capability scope for an agent.

Parameters:

  • agent_id (str)

  • permitted_tools (list[str])

  • forbidden_tools (list[str])

  • allowed_data_sources (list[str])

  • permitted_endpoints (list[str])

  • max_output_tokens (int)

  • external_contact (bool)

  • can_spawn_agents (bool)

  • max_delegation_depth (int)

agent_id: str
allowed_data_sources: list[str]
attenuate(other)[source]

Return monotonic attenuation: intersection of both manifests.

Parameters:

other (CapabilityManifest) – Sub-agent manifest to attenuate against this manifest.

Returns:

New manifest with reduced (never expanded) permissions.

Return type:

CapabilityManifest

can_spawn_agents: bool = False
external_contact: bool = False
forbidden_tools: list[str]
classmethod from_dict(data)[source]

Build a manifest from an already-validated dictionary.

Parameters:

data (dict[str, Any])

Return type:

CapabilityManifest

classmethod from_yaml(path)[source]

Load and validate a manifest from a YAML file.

Parameters:

path (str) – Path to the YAML manifest file.

Returns:

Validated CapabilityManifest instance.

Raises:

  • jsonschema.ValidationError – If validation fails.

  • ValueError – If YAML content is invalid.

Return type:

CapabilityManifest

is_data_source_allowed(data_source)[source]

Return True if the data source is explicitly listed.

An empty allowed_data_sources list is deny-by-default so that attenuation intersections cannot accidentally expand access.

Parameters:

data_source (str)

Return type:

bool

is_endpoint_permitted(endpoint)[source]

Return True if external contact is allowed and the endpoint matches.

Matching is URL-aware: scheme, host, and port must match an allow-list entry exactly; the path is matched as a segment-safe prefix (/v1 permits /v1/reports but not /v1evil).

Parameters:

endpoint (str)

Return type:

bool

is_tool_permitted(tool_name)[source]

Return True if the tool is explicitly permitted and not forbidden.

Parameters:

tool_name (str)

Return type:

bool

max_delegation_depth: int = 0
max_output_tokens: int = 4096
permitted_endpoints: list[str]
permitted_tools: list[str]

Firewall

AgentGuard firewall — public entry point.

class agentguard.firewall.AgentGuard(risk_threshold=0.85, enable_trust_attestation=True, enable_capability_enforcement=True, audit_log_path='./audit.jsonl', mode='enforce', task_objective=None, enable_consistency_check=True, consistency_threshold=0.1, consistency_ml_risk_floor=0.75, model_path=None, require_ml_model=False, enable_otel_export=False, include_message_preview=True, max_message_bytes=1048576)[source]

Bases: object

Inter-agent security middleware entry point.

Parameters:

  • risk_threshold (float)

  • enable_trust_attestation (bool)

  • enable_capability_enforcement (bool)

  • audit_log_path (str | None)

  • mode (str)

  • task_objective (str | None)

  • enable_consistency_check (bool)

  • consistency_threshold (float)

  • consistency_ml_risk_floor (float)

  • model_path (str | None)

  • require_ml_model (bool)

  • enable_otel_export (bool)

  • include_message_preview (bool)

  • max_message_bytes (int)

check_data_source(agent_id, data_source)[source]

Validate data-source access; False when blocked in enforce mode.

Parameters:

  • agent_id (str)

  • data_source (str)

Return type:

bool

check_endpoint(agent_id, endpoint)[source]

Validate an external endpoint; False when blocked in enforce mode.

Parameters:

  • agent_id (str)

  • endpoint (str)

Return type:

bool

check_output_tokens(agent_id, token_count)[source]

Validate output size; False when blocked in enforce mode.

Parameters:

  • agent_id (str)

  • token_count (int)

Return type:

bool

check_tool_call(agent_id, tool_name, *, endpoint=None)[source]

Validate a tool call; False when blocked in enforce mode.

Parameters:

  • agent_id (str)

  • tool_name (str)

  • endpoint (str | None)

Return type:

bool

get_manifest(agent_id)[source]

Return the registered capability manifest for an agent, if any.

Parameters:

agent_id (str)

Return type:

CapabilityManifest | None

inspect_content(sender_id, recipient_id, message)[source]

Inspect unsigned content (rules + ML + consistency, no trust check).

Use this at boundaries where messages cannot carry Ed25519 signatures — user input, framework middleware hooks, or third-party content. The decision is audited like any other inspection.

Parameters:

  • sender_id (str)

  • recipient_id (str)

  • message (str)

Return type:

FirewallDecision

inspect_message(sender_id, recipient_id, message, payload_bytes, *, signature=None)[source]

Run inspection, trust, and size checks on a message.

Parameters:

  • sender_id (str)

  • recipient_id (str)

  • message (str)

  • payload_bytes (bytes)

  • signature (bytes | None)

Return type:

FirewallDecision

inspect_tool_output(tool_name, output, agent_id)[source]

Inspect a tool return value for poisoning; audits the decision.

Parameters:

  • tool_name (str)

  • output (str)

  • agent_id (str)

Return type:

FirewallDecision

property is_ml_model_loaded: bool

Return True when the DeBERTa ONNX scorer is active.

register_agent(agent_id, manifest)[source]

Register agent identity, trust keys, and capability manifest.

Parameters:
Return type:

None

register_delegated_agent(parent_agent_id, child_manifest)[source]

Register a sub-agent with monotonic capability attenuation.

Parameters:
Return type:

CapabilityManifest

rotate_keys()[source]

Rotate ephemeral trust keys for all registered agents.

Return type:

None

sign_payload(sender_id, payload_bytes, *, recipient_id, nonce=None)[source]

Sign an inter-agent envelope (sender, recipient, nonce, payload).

The returned wire signature binds the recipient and a unique nonce so a previously signed payload cannot be redirected or silently replayed. Pass the bytes unchanged to inspect_message() as signature.

Parameters:
Return type:

bytes

wrap(graph)[source]

Wrap a LangGraph compiled graph with AgentGuard interception.

Parameters:

graph (object)

Return type:

object

wrap_mcp_tool(tool_fn, agent_id)[source]

Wrap a sync MCP tool callable with output inspection.

Parameters:

  • tool_fn (F)

  • agent_id (str)

Return type:

F

wrap_mcp_tool_async(tool_fn, agent_id)[source]

Wrap an async MCP tool callable with output inspection.

Parameters:

  • tool_fn (AF)

  • agent_id (str)

Return type:

AF

class agentguard.firewall.FirewallDecision(action, risk_score, trust_result, capability_result, failure_reason)[source]

Bases: object

Outcome of inspecting an inter-agent message.

Parameters:

  • action (str)

  • risk_score (float)

  • trust_result (str)

  • capability_result (str)

  • failure_reason (str | None)

action: str
capability_result: str
failure_reason: str | None
risk_score: float
trust_result: str

Capability manifests

YAML capability manifest loading and validation.

class agentguard.capability.manifest.CapabilityManifest(agent_id, permitted_tools=<factory>, forbidden_tools=<factory>, allowed_data_sources=<factory>, permitted_endpoints=<factory>, max_output_tokens=4096, external_contact=False, can_spawn_agents=False, max_delegation_depth=0)[source]

Bases: object

Declarative runtime capability scope for an agent.

Parameters:

  • agent_id (str)

  • permitted_tools (list[str])

  • forbidden_tools (list[str])

  • allowed_data_sources (list[str])

  • permitted_endpoints (list[str])

  • max_output_tokens (int)

  • external_contact (bool)

  • can_spawn_agents (bool)

  • max_delegation_depth (int)

agent_id: str
allowed_data_sources: list[str]
attenuate(other)[source]

Return monotonic attenuation: intersection of both manifests.

Parameters:

other (CapabilityManifest) – Sub-agent manifest to attenuate against this manifest.

Returns:

New manifest with reduced (never expanded) permissions.

Return type:

CapabilityManifest

can_spawn_agents: bool = False
external_contact: bool = False
forbidden_tools: list[str]
classmethod from_dict(data)[source]

Build a manifest from an already-validated dictionary.

Parameters:

data (dict[str, Any])

Return type:

CapabilityManifest

classmethod from_yaml(path)[source]

Load and validate a manifest from a YAML file.

Parameters:

path (str) – Path to the YAML manifest file.

Returns:

Validated CapabilityManifest instance.

Raises:

  • jsonschema.ValidationError – If validation fails.

  • ValueError – If YAML content is invalid.

Return type:

CapabilityManifest

is_data_source_allowed(data_source)[source]

Return True if the data source is explicitly listed.

An empty allowed_data_sources list is deny-by-default so that attenuation intersections cannot accidentally expand access.

Parameters:

data_source (str)

Return type:

bool

is_endpoint_permitted(endpoint)[source]

Return True if external contact is allowed and the endpoint matches.

Matching is URL-aware: scheme, host, and port must match an allow-list entry exactly; the path is matched as a segment-safe prefix (/v1 permits /v1/reports but not /v1evil).

Parameters:

endpoint (str)

Return type:

bool

is_tool_permitted(tool_name)[source]

Return True if the tool is explicitly permitted and not forbidden.

Parameters:

tool_name (str)

Return type:

bool

max_delegation_depth: int = 0
max_output_tokens: int = 4096
permitted_endpoints: list[str]
permitted_tools: list[str]

ML scorer

Production ML risk scorer using ONNX DeBERTa model.

class agentguard.inspector.ml_scorer.MLRiskScorer[source]

Bases: object

ONNX inference wrapper for inter-agent injection risk scoring.

is_model_loaded()[source]

Return True when the ONNX session is active.

Return type:

bool

load_model(model_path)[source]

Load and verify ONNX model and tokenizer.

Parameters:

model_path (str)

Return type:

None

property model_path: str | None

Return loaded model path or None.

score(message)[source]

Return injection probability in [0.0, 1.0].

Parameters:

message (str)

Return type:

float

exception agentguard.inspector.ml_scorer.ModelIntegrityError[source]

Bases: Exception

Raised when ONNX model hash does not match the pinned value.

exception agentguard.inspector.ml_scorer.ModelNotLoadedWarning[source]

Bases: UserWarning

Emitted when score() is called without a loaded model.

MCP output inspection

MCP tool output inspection before agent ingestion.

class agentguard.mcp.output_inspector.MCPInspectionResult(safe, risk_score, flagged_patterns, tool_name, calling_agent_id)[source]

Bases: object

Result of inspecting an MCP tool return value.

Parameters:
calling_agent_id: str
flagged_patterns: list[str]
risk_score: float
safe: bool
tool_name: str
class agentguard.mcp.output_inspector.MCPOutputInspector(rule_filter, ml_scorer, threshold=0.85)[source]

Bases: object

Inspects MCP server tool outputs for injection payloads.

Parameters:
inspect_output(tool_name, output, calling_agent_id)[source]

Score MCP tool output for injection risk.

Parameters:

  • tool_name (str)

  • output (str)

  • calling_agent_id (str)

Return type:

MCPInspectionResult

wrap_tool(tool_fn, guard, agent_id)[source]

Wrap a synchronous tool callable with audited MCP output inspection.

Parameters:

  • tool_fn (F)

  • guard (Any)

  • agent_id (str)

Return type:

F

wrap_tool_async(tool_fn, guard, agent_id)[source]

Wrap an async tool callable with audited MCP output inspection.

Parameters:

  • tool_fn (AF)

  • guard (Any)

  • agent_id (str)

Return type:

AF

exception agentguard.mcp.output_inspector.MCPPoisoningException(*, tool_name, agent_id, risk_score, flagged_patterns, action='QUARANTINE', sender_id='', recipient_id='', trust_result='SKIP', capability_result='SKIP', failure_reason=None)[source]

Bases: AgentGuardException

Raised when MCP tool output contains an injection payload.

Parameters:

  • tool_name (str)

  • agent_id (str)

  • risk_score (float)

  • flagged_patterns (list[str])

  • action (str)

  • sender_id (str)

  • recipient_id (str)

  • trust_result (str)

  • capability_result (str)

  • failure_reason (str | None)

Return type:

None