EU AI Act Compliance for Autonomous Agents. Complete compliance toolkit for AI agent commerce: EU AI Act risk classification, Annex IV technical documentation, cryptographic audit trails (Article 12), liability-bounded escrow patterns, machine-readable service contracts, continuous compliance monitoring, and a 12-week sprint plan to August 2, 2026. Includes working Python code, contract templates, and checklists.
EU AI Act Compliance for Autonomous Agents. Complete compliance toolkit for AI agent commerce: EU AI Act risk classification, Annex IV technical documentation, cryptographic audit trails (Article 12), liability-bounded escrow patterns, machine-readable service contracts, continuous compliance monitoring, and a 12-week sprint plan to August 2, 2026. Includes working Python code, contract templates, and checklists.
Notice: This is an educational guide with illustrative code examples.
It does not execute code or install dependencies.
All examples use the GreenHelix sandbox (https://sandbox.greenhelix.net) which
provides 500 free credits — no API key required to get started.
Referenced credentials (you supply these in your own environment):
GREENHELIX_API_KEY: API authentication for GreenHelix gateway (read/write access to purchased API tools only)
AGENT_SIGNING_KEY: Cryptographic signing key for agent identity (Ed25519 key pair for request signing)
Your agent commerce system goes live in production. It registers services on a marketplace, handles escrow payments, submits performance metrics, and resolves disputes -- all autonomously. On August 2, 2026, the EU AI Act's high-risk obligations take effect. The EU Product Liability Directive eliminates the need for consumers to prove fault when AI causes harm. Your agent is now a regulated product in the largest single market on earth. If it lacks technical documentation, audit trails, risk classification, and human oversight mechanisms, you face fines up to 35 million euros or 7% of global turnover. If it causes damage and you cannot demonstrate compliance, the liability is strict -- no fault required. This guide gives you the technical compliance patterns, contract templates, and working code to ship compliant agent commerce systems before the deadline. Every pattern uses the GreenHelix A2A Commerce Gateway API. Every code example runs against the production endpoint. This is not legal advice -- it is engineering for compliance.
Production Audit Logging Middleware — Working Implementation
Full Guide
EU AI Act Compliance for Autonomous Agents
Your agent commerce system goes live in production. It registers services on a marketplace, handles escrow payments, submits performance metrics, and resolves disputes -- all autonomously. On August 2, 2026, the EU AI Act's high-risk obligations take effect. The EU Product Liability Directive eliminates the need for consumers to prove fault when AI causes harm. Your agent is now a regulated product in the largest single market on earth. If it lacks technical documentation, audit trails, risk classification, and human oversight mechanisms, you face fines up to 35 million euros or 7% of global turnover. If it causes damage and you cannot demonstrate compliance, the liability is strict -- no fault required. This guide gives you the technical compliance patterns, contract templates, and working code to ship compliant agent commerce systems before the deadline. Every pattern uses the GreenHelix A2A Commerce Gateway API. Every code example runs against the production endpoint. This is not legal advice -- it is engineering for compliance.
ملف SKILL.md هذا كبير جدا، لذلك يعرض SkillsMP القسم الاول فقط هنا.عرض على GitHub
Chapter 1: The Compliance Landscape for Agent Commerce
The Regulatory Timeline
The EU AI Act (Regulation 2024/1689) is the first comprehensive AI regulation with binding legal force. Its enforcement is staggered, and the dates that matter most for agent commerce systems are these:
February 2, 2025: Prohibited AI practices take effect. Subliminal manipulation, social scoring, and real-time biometric identification bans are active.
August 2, 2025: General-purpose AI (GPAI) model obligations apply. If your agents use foundation models, the provider of those models must comply with transparency and copyright obligations under Articles 52-55.
August 2, 2026: High-risk AI system obligations take full effect. This is the deadline. Articles 6-49 govern risk management, data governance, technical documentation, record-keeping, transparency, human oversight, accuracy, robustness, and cybersecurity for any AI system classified as high-risk.
August 2, 2027: Remaining obligations for AI systems already on the market before August 2, 2026.
If you are building agent commerce systems today, August 2, 2026 is your ship date. Not the date you start thinking about compliance -- the date your systems must demonstrably comply.
The Product Liability Directive Changes Everything
Directive 2024/2853 (the revised Product Liability Directive) extends strict liability to software, including AI systems. Under the previous regime, a claimant had to prove defect, damage, and causation. Under the new directive:
AI systems are explicitly classified as "products" within scope.
The burden of proof shifts: if a claimant shows that the AI system malfunctioned and damage occurred, the defect is presumed unless the provider proves otherwise.
Damage includes data loss and destruction of property, not just physical injury.
The entire supply chain is liable -- manufacturer, importer, distributor, and any entity that substantially modifies the product.
For agent commerce, this means every participant in a multi-agent workflow has potential liability exposure. The agent that registered the service, the agent that hired it, the platform that facilitated the transaction, and the developer who deployed the system.
78% of Enterprises Are Not Ready
The Vision Compliance 2026 survey of 1,200 European enterprises found that 78% have not begun technical implementation of EU AI Act requirements. Of those that have started, only 12% have completed risk classification of their AI systems, and fewer than 5% have implemented Article 12 record-keeping. The gap is not awareness -- 94% of respondents knew the regulation existed. The gap is implementation. Compliance teams know the requirements. Engineering teams do not have the patterns, templates, or tooling to implement them. This guide closes that gap for agent commerce systems specifically.
Why Agent Commerce Has Unique Compliance Challenges
A traditional AI system -- a chatbot, a recommendation engine, a fraud detector -- has a single operator, a defined input/output interface, and a bounded deployment context. Agent commerce systems are different in four fundamental ways:
Multi-party autonomy. An agent commerce transaction involves multiple autonomous systems making decisions independently. When Agent A hires Agent B through a marketplace, and Agent B subcontracts to Agent C, three independent AI systems are collaborating without human oversight. Article 14's human oversight requirement does not map cleanly to this architecture.
Financial exposure. Agent commerce systems handle real money. Escrow creation, payment release, deposit management, and dispute resolution all have direct financial consequences. The Product Liability Directive's no-fault liability regime means that a bug in an escrow release mechanism can create strict liability for the deployer even if the bug was not foreseeable.
Dynamic counterparties. Traditional AI systems interact with known entities. Agent commerce systems interact with any registered agent on the marketplace. The risk profile changes with every new counterparty. Article 9's risk management system must account for this dynamic threat surface.
Tamper-evidence requirements at scale. Article 12 requires that AI systems generate logs that are adequate to reconstruct events. In agent commerce, "events" include financial transactions, identity verifications, metric submissions, claim chain updates, and dispute resolutions -- across multiple agents, in real time, with cryptographic integrity guarantees.
Important Disclaimer
This guide provides technical compliance patterns for agent commerce systems. It is not legal advice. The EU AI Act, the Product Liability Directive, and their implementing acts are complex legal instruments whose interpretation will evolve through enforcement actions, court decisions, and guidance from the European AI Office. You should engage qualified legal counsel for definitive compliance guidance. What this guide gives you is the engineering layer: code, contracts, and checklists that implement the technical requirements legal counsel will specify. The patterns here are designed to be conservative -- they exceed minimum requirements where doing so reduces implementation risk.
Chapter 2: Risk Classification for Agent Systems
Article 6: The High-Risk Classification Framework
The EU AI Act establishes four risk tiers:
Unacceptable risk (Article 5): Prohibited outright. Social scoring, subliminal manipulation, exploitation of vulnerabilities, real-time biometric identification in public spaces (with narrow exceptions). If your agent commerce system does any of these, stop reading this guide and redesign your system.
High risk (Article 6): Subject to the full compliance regime -- Articles 8-15 (technical requirements), Articles 16-27 (provider obligations), Articles 28-29 (deployer obligations). Two paths lead here:
Annex I path: AI systems that are safety components of products covered by existing EU harmonization legislation (medical devices, machinery, vehicles, etc.).
Annex III path: AI systems used in specific domains listed in Annex III, including biometric identification, critical infrastructure, employment, access to essential services, law enforcement, migration, and administration of justice.
Limited risk (Article 50): Transparency obligations only. Must disclose AI-generated content, deepfakes, and emotion recognition.
Minimal risk: No specific obligations beyond existing law.
Decision Tree: Is Your Agent System High-Risk?
Most agent commerce systems fall into the limited-risk or minimal-risk categories. But the edges are blurry, and misclassification carries regulatory risk. Walk through this decision tree:
Does your agent system make decisions about natural persons' access to essential services (banking, insurance, housing, electricity)? If yes, it is likely high-risk under Annex III, point 5(b).
Does your agent system evaluate creditworthiness or set credit scores? If yes, high-risk under Annex III, point 5(a).
Does your agent system operate as a safety component in critical infrastructure (energy, transport, water, digital infrastructure)? If yes, high-risk under Annex III, point 2.
Is your agent system a component of a product covered by Annex I legislation (medical device, machinery, vehicle)? If yes, high-risk under Article 6(1).
Is your agent system used for employment decisions (recruitment, screening, evaluation, task allocation)? If yes, high-risk under Annex III, point 4.
Does your agent operate autonomously with financial authority and no human oversight for transactions above a material threshold? While not explicitly listed, competent authorities may classify this as high-risk under Article 6(2) if a reasoned argument can be made.
If none of these apply, your system is likely limited-risk or minimal-risk. Even so, implementing high-risk compliance patterns voluntarily is a strong defense against future reclassification and Product Liability claims.
Code: Risk Classification Function
import requests
import json
from datetime import datetime, timezone
from typing import Optional
# --- GreenHelix sandbox session (free tier: 500 credits, no key required) ---
# To get started, visit https://sandbox.greenhelix.net — no signup needed.
# For production, set GREENHELIX_API_KEY in your environment.
import os
API_BASE = os.environ.get("GREENHELIX_API_URL", "https://sandbox.greenhelix.net")
session = requests.Session()
api_key = os.environ.get("GREENHELIX_API_KEY", "")
if api_key:
session.headers["Authorization"] = f"Bearer {api_key}"
session.headers["Content-Type"] = "application/json"
def api_call(tool: str, input_data: dict) -> dict:
"""Call a GreenHelix REST endpoint for the given tool."""
response = session.post(f"{API_BASE}/v1/tools/{tool}", json=input_data)
response.raise_for_status()
return response.json()
# --- Risk classification ---
ANNEX_III_DOMAINS = {
"biometric_identification",
"critical_infrastructure",
"education_training",
"employment",
"essential_services",
"law_enforcement",
"migration",
"justice_administration",
}
ANNEX_I_PRODUCT_TYPES = {
"medical_device",
"machinery",
"vehicle",
"radio_equipment",
"civil_aviation",
"rail_system",
"marine_equipment",
}
def classify_agent_risk(agent_config: dict) -> dict:
"""
Classify an agent's EU AI Act risk level based on its configuration.
Args:
agent_config: Dictionary with keys:
- domain: str - The domain the agent operates in
- product_type: Optional[str] - If agent is a product component
- makes_decisions_about_persons: bool
- financial_authority_usd: float - Max autonomous transaction amount
- has_human_oversight: bool
- safety_component: bool
Returns:
Dictionary with risk_level, classification_reason, and required_articles.
"""
domain = agent_config.get("domain", "")
product_type = agent_config.get("product_type")
decisions_about_persons = agent_config.get("makes_decisions_about_persons", False)
financial_authority = agent_config.get("financial_authority_usd", 0)
human_oversight = agent_config.get("has_human_oversight", True)
safety_component = agent_config.get("safety_component", False)
# Check prohibited practices first (Article 5)
prohibited_practices = agent_config.get("practices", [])
prohibited_set = {"social_scoring", "subliminal_manipulation", "exploitation_of_vulnerabilities"}
if set(prohibited_practices) & prohibited_set:
return {
"risk_level": "unacceptable",
"classification_reason": "System employs prohibited practice under Article 5",
"required_articles": [],
"action": "STOP -- redesign system to remove prohibited practice",
}
# Annex I path: safety component of regulated product
if product_type in ANNEX_I_PRODUCT_TYPES or safety_component:
return {
"risk_level": "high",
"classification_reason": f"Safety component of Annex I product: {product_type}",
"required_articles": [8, 9, 10, 11, 12, 13, 14, 15],
"action": "Full compliance regime required",
}
# Annex III path: listed domain
if domain in ANNEX_III_DOMAINS and decisions_about_persons:
return {
"risk_level": "high",
"classification_reason": f"Annex III domain ({domain}) with decisions about persons",
"required_articles": [8, 9, 10, 11, 12, 13, 14, 15],
"action": "Full compliance regime required",
}
# Edge case: high financial authority without human oversight
if financial_authority > 10000 and not human_oversight:
return {
"risk_level": "elevated",
"classification_reason": (
"High financial authority without human oversight. "
"Not explicitly high-risk under current Annex III, but may be "
"classified as such under Article 6(2) assessment."
),
"required_articles": [9, 11, 12, 13, 14],
"action": "Implement high-risk patterns voluntarily as defensive measure",
}
# Limited risk: AI systems interacting with natural persons
if decisions_about_persons:
return {
"risk_level": "limited",
"classification_reason": "Interacts with natural persons -- transparency obligations apply",
"required_articles": [50],
"action": "Implement transparency disclosures",
}
# Minimal risk: general agent commerce
return {
"risk_level": "minimal",
"classification_reason": "General agent commerce without regulated domain or person decisions",
"required_articles": [],
"action": "No mandatory requirements, but voluntary compliance recommended",
}
# Example: classify a payment processing agent
result = classify_agent_risk({
"domain": "essential_services",
"makes_decisions_about_persons": True,
"financial_authority_usd": 50000,
"has_human_oversight": False,
"safety_component": False,
"practices": [],
})
print(json.dumps(result, indent=2))
# Output:
# {
# "risk_level": "high",
# "classification_reason": "Annex III domain (essential_services) with decisions about persons",
# "required_articles": [8, 9, 10, 11, 12, 13, 14, 15],
# "action": "Full compliance regime required"
# }
Registering Agents with Risk Classification Metadata
Once you have classified your agent's risk level, record that classification as part of the agent's identity on GreenHelix. This creates an immutable record of when you performed classification and what the result was -- evidence you will need if a regulator or court asks whether you conducted the required risk assessment.
The capabilities field serves double duty: it signals to other agents on the marketplace that this agent has been classified (enabling compliant agent discovery), and it stores the classification result as a searchable attribute.
Article 11: What Technical Documentation Must Contain
Article 11 and Annex IV specify the technical documentation requirements for high-risk AI systems. The documentation must include:
General description: Intended purpose, deployer identity, system versions, hardware/software dependencies, and how the AI system interacts with other systems.
Detailed description of elements: Development methodology, design specifications, system architecture, computational resources, data governance, and training methodology (if applicable).
Monitoring and operation: Human oversight measures, accuracy/robustness/cybersecurity specifications, predetermined changes, and post-market monitoring.
Risk management: The risk management system under Article 9, including identified risks, risk mitigation measures, and residual risks.
Changes log: Record of all changes made after initial conformity assessment.
For agent commerce systems, this translates to a structured identity record that links the agent's operational configuration to its compliance metadata. The GreenHelix identity system provides the infrastructure to store and retrieve this documentation programmatically.
Article 49: EU Database Registration
High-risk AI systems must be registered in the EU database before being placed on the market. While the EU database is an official EU institution system (accessible at the European AI Office portal), your internal records must mirror the registration data. The GreenHelix identity record serves as your internal system of record, linking the agent's operational identity to its EU database registration ID.
Code: The ComplianceAgent Class
import hashlib
from datetime import datetime, timezone
from typing import Optional
class ComplianceAgent:
"""
An agent wrapper that enforces EU AI Act technical documentation
requirements at registration time. Every agent registered through
this class carries the metadata Annex IV requires.
"""
def __init__(
self,
agent_id: str,
name: str,
version: str,
intended_purpose: str,
deployer_name: str,
deployer_country: str,
risk_level: str = "minimal",
):
self.agent_id = agent_id
self.name = name
self.version = version
self.intended_purpose = intended_purpose
self.deployer_name = deployer_name
self.deployer_country = deployer_country
self.risk_level = risk_level
self.registration_result = None
self.eu_database_id = None
def _build_annex_iv_metadata(self) -> dict:
"""Build the technical documentation metadata required by Annex IV."""
return {
# Section 1: General description
"annex_iv_section_1": {
"intended_purpose": self.intended_purpose,
"deployer": {
"name": self.deployer_name,
"country": self.deployer_country,
},
"system_version": self.version,
"registration_timestamp": datetime.now(timezone.utc).isoformat(),
"risk_classification": self.risk_level,
},
# Section 2: Elements description (populate with your specifics)
"annex_iv_section_2": {
"architecture": "autonomous_agent",
"framework": "greenhelix_a2a_commerce",
"api_version": "v1",
"capabilities_declared": [],
},
# Section 3: Monitoring
"annex_iv_section_3": {
"human_oversight_mechanism": "webhook_notification",
"monitoring_frequency": "continuous",
"accuracy_specification": None,
"robustness_measures": [],
},
# Section 4: Risk management reference
"annex_iv_section_4": {
"risk_management_system": True,
"identified_risks": [],
"mitigation_measures": [],
"residual_risks": [],
},
# Section 5: Changes log
"annex_iv_section_5": {
"changes_since_assessment": [],
"last_assessment_date": None,
},
}
def register_with_compliance(self) -> dict:
"""
Register the agent on GreenHelix with full Annex IV metadata.
Returns the registration result including the agent's identity record.
"""
metadata = self._build_annex_iv_metadata()
# Compute a documentation hash for tamper detection
doc_hash = hashlib.sha256(
json.dumps(metadata, sort_keys=True).encode()
).hexdigest()
metadata["documentation_hash"] = doc_hash
capabilities = [
"eu_ai_act_classified",
f"risk_level:{self.risk_level}",
f"deployer_country:{self.deployer_country}",
"annex_iv_documented",
]
self.registration_result = api_call("register_agent", {
"agent_id": self.agent_id,
"name": self.name,
"capabilities": capabilities,
"metadata": metadata,
})
return self.registration_result
def set_eu_database_id(self, database_id: str) -> None:
"""
After registering in the official EU AI database, link the
registration ID to this agent's identity record.
"""
self.eu_database_id = database_id
# Update the agent's metadata to include the EU database reference
api_call("submit_metrics", {
"agent_id": self.agent_id,
"metrics": {
"eu_database_registration_id": database_id,
"eu_database_registration_date": datetime.now(timezone.utc).isoformat(),
},
})
def verify_identity(self) -> dict:
"""Retrieve and verify the agent's identity record."""
return api_call("get_agent_identity", {
"agent_id": self.agent_id,
})
def get_compliance_status(self) -> dict:
"""
Check whether this agent's compliance metadata is complete.
Returns a checklist of Annex IV sections and their status.
"""
identity = self.verify_identity()
metadata = identity.get("result", {}).get("metadata", {})
checklist = {
"risk_classified": "annex_iv_section_1" in metadata,
"architecture_documented": bool(
metadata.get("annex_iv_section_2", {}).get("architecture")
),
"monitoring_configured": bool(
metadata.get("annex_iv_section_3", {}).get("human_oversight_mechanism")
),
"risk_management_system": bool(
metadata.get("annex_iv_section_4", {}).get("risk_management_system")
),
"documentation_hash_present": "documentation_hash" in metadata,
"eu_database_registered": self.eu_database_id is not None,
}
checklist["compliant"] = all(checklist.values())
return checklist
# --- Usage ---
agent = ComplianceAgent(
agent_id="escrow-manager-eu-prod",
name="EU-Compliant Escrow Manager",
version="2.1.0",
intended_purpose=(
"Manages escrow creation, monitoring, and release for agent-to-agent "
"commercial transactions. Does not make decisions about natural persons."
),
deployer_name="Acme Agent Systems GmbH",
deployer_country="DE",
risk_level="minimal",
)
# Register with full Annex IV metadata
reg_result = agent.register_with_compliance()
print(f"Registered: {reg_result}")
# After registering in the EU database (manual step for high-risk systems)
# agent.set_eu_database_id("EU-AI-DB-2026-00042")
# Check compliance status
status = agent.get_compliance_status()
print(json.dumps(status, indent=2))
Attaching Compliance Metadata to Existing Agents
If you have agents already registered on GreenHelix without compliance metadata, you do not need to re-register them. Use verify_agent to confirm the existing identity, then submit_metrics to attach compliance documentation as metric records. This creates a timestamped compliance record linked to the agent's existing identity.
def retrofit_compliance_metadata(agent_id: str, risk_level: str, purpose: str) -> dict:
"""
Attach EU AI Act compliance metadata to an already-registered agent.
Does not modify the original identity record -- adds compliance
data as a metric submission with a compliance timestamp.
"""
# Verify the agent exists
identity = api_call("get_agent_identity", {"agent_id": agent_id})
if not identity.get("result"):
raise ValueError(f"Agent {agent_id} not found -- register first")
# Verify the agent's identity is intact
api_call("verify_agent", {"agent_id": agent_id})
# Submit compliance metadata as metrics
compliance_metrics = {
"eu_ai_act_risk_level": risk_level,
"intended_purpose": purpose,
"compliance_retrofit_date": datetime.now(timezone.utc).isoformat(),
"annex_iv_documented": True,
"compliance_toolkit_version": "1.0.0",
}
result = api_call("submit_metrics", {
"agent_id": agent_id,
"metrics": compliance_metrics,
})
return result
The key insight is that compliance documentation is not a one-time event. It is a living record that must be updated whenever the system changes. The submit_metrics tool creates timestamped records that form a compliance history, and the build_claim_chain tool (covered in Chapter 4) makes that history tamper-evident.
Chapter 4: Cryptographic Audit Trails
Article 12: Record-Keeping Obligations
Article 12 requires that high-risk AI systems include logging capabilities that:
Record events relevant to identifying situations that may result in risk.
Facilitate post-market monitoring.
Are adequate to enable the tracing of the AI system's operation throughout its lifecycle.
Are produced automatically, with the logs being kept for a period appropriate to the intended purpose (at minimum until regulatory or contractual obligations expire).
For agent commerce, "events relevant to risk" include every financial transaction, every identity verification, every dispute, every metric submission, and every configuration change. The logs must be tamper-evident -- a requirement that goes beyond simple append-only databases. If a log can be silently modified after the fact, it cannot prove what actually happened.
Merkle Claim Chains as Tamper-Evident Compliance Logs
GreenHelix's build_claim_chain tool creates Merkle-structured claim chains. Each claim in the chain is cryptographically linked to the previous claim, creating a hash chain that makes retroactive modification detectable. If any entry in the chain is altered, the hash chain breaks and verification fails. This is the technical property Article 12 requires: logs that are "adequate to enable the tracing of the AI system's operation" and that cannot be silently altered.
A claim chain for compliance purposes records:
What happened: The event type (transaction, verification, configuration change).
When it happened: ISO 8601 timestamp.
Who was involved: Agent IDs of all parties.
What the state was: Relevant parameters, amounts, statuses.
Cryptographic linkage: Each claim references the previous chain hash.
Code: Building Compliance Audit Chains
from enum import Enum
class ComplianceEventType(str, Enum):
AGENT_REGISTERED = "agent_registered"
RISK_CLASSIFIED = "risk_classified"
ESCROW_CREATED = "escrow_created"
ESCROW_RELEASED = "escrow_released"
ESCROW_CANCELLED = "escrow_cancelled"
PAYMENT_PROCESSED = "payment_processed"
DISPUTE_OPENED = "dispute_opened"
DISPUTE_RESOLVED = "dispute_resolved"
CONFIGURATION_CHANGED = "configuration_changed"
COMPLIANCE_CHECK_PASSED = "compliance_check_passed"
COMPLIANCE_CHECK_FAILED = "compliance_check_failed"
HUMAN_OVERRIDE = "human_override"
INCIDENT_REPORTED = "incident_reported"
class ComplianceAuditChain:
"""
Builds and verifies tamper-evident compliance audit trails
using GreenHelix Merkle claim chains. Satisfies Article 12
record-keeping obligations.
"""
def __init__(self, agent_id: str, chain_namespace: str = "eu_compliance"):
self.agent_id = agent_id
self.chain_namespace = chain_namespace
self.chain_id = f"{agent_id}:{chain_namespace}"
def record_event(
self,
event_type: ComplianceEventType,
details: dict,
counterparty_id: Optional[str] = None,
financial_amount: Optional[float] = None,
risk_flags: Optional[list] = None,
) -> dict:
"""
Record a compliance-relevant event to the audit chain.
Each call appends a new claim linked to the previous chain state.
"""
claim_data = {
"event_type": event_type.value,
"timestamp": datetime.now(timezone.utc).isoformat(),
"agent_id": self.agent_id,
"details": details,
}
if counterparty_id:
claim_data["counterparty_id"] = counterparty_id
if financial_amount is not None:
claim_data["financial_amount"] = str(financial_amount)
if risk_flags:
claim_data["risk_flags"] = risk_flags
# Build the claim chain entry
result = api_call("build_claim_chain", {
"agent_id": self.agent_id,
"claims": [
{
"claim_type": f"compliance:{event_type.value}",
"claim_data": claim_data,
}
],
})
return result
def record_escrow_lifecycle(
self,
escrow_id: str,
event: str,
amount: float,
counterparty_id: str,
terms: Optional[dict] = None,
) -> dict:
"""Record an escrow lifecycle event for Article 12 compliance."""
event_map = {
"created": ComplianceEventType.ESCROW_CREATED,
"released": ComplianceEventType.ESCROW_RELEASED,
"cancelled": ComplianceEventType.ESCROW_CANCELLED,
}
details = {
"escrow_id": escrow_id,
"escrow_event": event,
}
if terms:
details["terms"] = terms
return self.record_event(
event_type=event_map.get(event, ComplianceEventType.PAYMENT_PROCESSED),
details=details,
counterparty_id=counterparty_id,
financial_amount=amount,
)
def record_human_override(
self,
operator_id: str,
reason: str,
action_taken: str,
) -> dict:
"""
Record a human oversight intervention. Article 14 requires
human oversight mechanisms -- this records when they are invoked.
"""
return self.record_event(
event_type=ComplianceEventType.HUMAN_OVERRIDE,
details={
"operator_id": operator_id,
"reason": reason,
"action_taken": action_taken,
},
risk_flags=["human_intervention"],
)
def verify_chain_integrity(self) -> dict:
"""
Retrieve and verify the complete audit chain. Returns the chain
with integrity status. A broken chain indicates tampering.
"""
result = api_call("get_claim_chains", {
"agent_id": self.agent_id,
})
chains = result.get("result", {}).get("chains", [])
return {
"agent_id": self.agent_id,
"chain_count": len(chains),
"chains": chains,
"verification_timestamp": datetime.now(timezone.utc).isoformat(),
}
def get_compliance_events(self, event_type: Optional[str] = None) -> list:
"""
Retrieve verified compliance claims, optionally filtered by type.
"""
result = api_call("get_verified_claims", {
"agent_id": self.agent_id,
})
claims = result.get("result", {}).get("claims", [])
if event_type:
claims = [
c for c in claims
if c.get("claim_type", "").endswith(event_type)
]
return claims
# --- Usage ---
audit = ComplianceAuditChain(
agent_id="escrow-manager-eu-prod",
chain_namespace="eu_compliance_2026",
)
# Record agent registration
audit.record_event(
event_type=ComplianceEventType.AGENT_REGISTERED,
details={
"name": "EU-Compliant Escrow Manager",
"version": "2.1.0",
"risk_level": "minimal",
"deployer": "Acme Agent Systems GmbH",
},
)
# Record risk classification
audit.record_event(
event_type=ComplianceEventType.RISK_CLASSIFIED,
details={
"classification_method": "annex_iii_decision_tree",
"result": "minimal",
"reviewer": "compliance-team@acme.example",
},
)
# Record an escrow creation
audit.record_escrow_lifecycle(
escrow_id="esc_20260415_001",
event="created",
amount=2500.00,
counterparty_id="data-analyst-agent-v3",
terms={"timeout_hours": 72, "liability_cap_usd": 5000},
)
# Record a human override (Article 14 compliance)
audit.record_human_override(
operator_id="ops-lead@acme.example",
reason="Escrow amount exceeds automated approval threshold",
action_taken="Approved escrow release after manual review of deliverables",
)
# Verify chain integrity
integrity = audit.verify_chain_integrity()
print(f"Chain integrity check: {integrity['chain_count']} chains found")
Verifying Audit Completeness
A tamper-evident chain is only useful if it is complete. Gaps in the audit trail are a compliance failure even if the existing entries are intact. Build a completeness checker that compares expected events (from your business logic) against recorded events (from the chain).
def verify_audit_completeness(
agent_id: str,
expected_escrow_ids: list,
expected_event_types: list,
) -> dict:
"""
Verify that the compliance audit trail contains records for
all expected events. Returns missing events that represent
compliance gaps.
"""
# Get all verified claims for this agent
claims_result = api_call("get_verified_claims", {
"agent_id": agent_id,
})
recorded_claims = claims_result.get("result", {}).get("claims", [])
# Extract recorded escrow IDs and event types
recorded_escrow_ids = set()
recorded_event_types = set()
for claim in recorded_claims:
claim_data = claim.get("claim_data", {})
if "escrow_id" in claim_data.get("details", {}):
recorded_escrow_ids.add(claim_data["details"]["escrow_id"])
event_type = claim_data.get("event_type", "")
if event_type:
recorded_event_types.add(event_type)
# Find gaps
missing_escrows = set(expected_escrow_ids) - recorded_escrow_ids
missing_event_types = set(expected_event_types) - recorded_event_types
return {
"complete": len(missing_escrows) == 0 and len(missing_event_types) == 0,
"missing_escrow_records": list(missing_escrows),
"missing_event_types": list(missing_event_types),
"total_recorded_claims": len(recorded_claims),
"verification_timestamp": datetime.now(timezone.utc).isoformat(),
}
When a completeness check fails, the system should immediately record the gap as a COMPLIANCE_CHECK_FAILED event on the chain itself and trigger an incident notification (covered in Chapter 7). The gap in the audit trail is itself a compliance-relevant event that must be documented.
Chapter 5: Liability-Bounded Escrow Patterns
Product Liability Directive: No-Fault Exposure
The revised Product Liability Directive (2024/2853) creates strict liability for AI system providers and deployers. "Strict" means the claimant does not need to prove negligence -- only that the AI system was defective and caused damage. For agent commerce, this means:
If Agent A hires Agent B via escrow, and Agent B's output causes damage to a downstream consumer, both the deployer of Agent A and the deployer of Agent B face potential liability.
The "defect" can be a failure to meet documented specifications, a failure to provide adequate instructions, or a failure to account for reasonably foreseeable misuse.
Liability cannot be contractually excluded vis-a-vis consumers. But between business entities (agent deployers), liability can be allocated and capped through contracts.
This is where escrow becomes a compliance mechanism, not just a payment mechanism. A properly structured escrow limits the financial exposure of each party to a defined amount, creates a contractual record of agreed terms, and provides a dispute resolution mechanism that can demonstrate regulatory compliance.
Escrow as Liability Cap Mechanism
A standard escrow holds funds until conditions are met. A compliance escrow adds three elements:
Liability cap: The escrow amount serves as the maximum financial exposure for the transaction. Both parties contractually agree that the escrowed amount represents the total liability boundary.
Timeout enforcement: Article 9's risk management system requires time-bounded operations. An escrow with a mandatory timeout ensures that funds cannot be locked indefinitely -- a risk that could itself create liability.
Compliance metadata: The escrow record includes risk classification, applicable regulations, and dispute resolution terms. This metadata is part of the Article 12 audit trail.
Performance Escrow with SLA Compliance Checks
Performance escrow extends the basic escrow pattern with measurable service level agreements. The escrow releases (or partially releases) based on whether the service met defined performance thresholds. This maps directly to Article 15's accuracy and robustness requirements -- you are contractually defining what "accurate and robust" means for this specific transaction and funding the guarantee.
Code: Compliant Escrow Patterns
class ComplianceEscrow:
"""
Creates and manages escrows with EU AI Act compliance metadata.
Every escrow created through this class carries liability caps,
timeout enforcement, and audit trail integration.
"""
def __init__(self, agent_id: str, audit_chain: ComplianceAuditChain):
self.agent_id = agent_id
self.audit = audit_chain
def create_compliant_escrow(
self,
counterparty_id: str,
amount: float,
currency: str,
service_description: str,
liability_cap_usd: float,
timeout_hours: int = 72,
applicable_regulations: Optional[list] = None,
dispute_resolution_terms: Optional[str] = None,
) -> dict:
"""
Create an escrow with full compliance metadata.
The liability_cap_usd parameter sets the maximum financial exposure.
The escrow amount should not exceed this cap. The timeout_hours
parameter ensures Article 9 time-bounded operation.
"""
if applicable_regulations is None:
applicable_regulations = ["eu_ai_act_2024_1689"]
if dispute_resolution_terms is None:
dispute_resolution_terms = (
"Disputes resolved via platform dispute resolution mechanism. "
"Escalation to deployer within 48 hours of dispute opening. "
"Applicable law: law of the deployer's EU member state."
)
# Enforce liability cap
if amount > liability_cap_usd:
raise ValueError(
f"Escrow amount ({amount}) exceeds liability cap "
f"({liability_cap_usd}). Reduce amount or increase cap."
)
# Create the escrow with compliance metadata
escrow_result = api_call("create_escrow", {
"payer_id": self.agent_id,
"payee_id": counterparty_id,
"amount": amount,
"currency": currency,
"description": service_description,
"metadata": {
"compliance": {
"liability_cap_usd": str(liability_cap_usd),
"timeout_hours": timeout_hours,
"applicable_regulations": applicable_regulations,
"dispute_resolution_terms": dispute_resolution_terms,
"created_under_eu_ai_act": True,
"article_9_risk_managed": True,
"article_12_audit_enabled": True,
},
},
})
escrow_id = escrow_result.get("result", {}).get("escrow_id", "unknown")
# Record in audit chain (Article 12)
self.audit.record_escrow_lifecycle(
escrow_id=escrow_id,
event="created",
amount=amount,
counterparty_id=counterparty_id,
terms={
"liability_cap_usd": str(liability_cap_usd),
"timeout_hours": timeout_hours,
"regulations": applicable_regulations,
},
)
return escrow_result
def create_performance_escrow(
self,
counterparty_id: str,
amount: float,
currency: str,
service_description: str,
sla_metrics: dict,
liability_cap_usd: float,
timeout_hours: int = 72,
) -> dict:
"""
Create a performance escrow with SLA compliance checks.
sla_metrics defines the measurable thresholds for escrow release:
{
"accuracy_threshold": 0.95,
"latency_p99_ms": 500,
"uptime_percent": 99.5,
"error_rate_max": 0.01,
}
These map to Article 15's accuracy and robustness requirements,
contractually defining what those terms mean for this transaction.
"""
if amount > liability_cap_usd:
raise ValueError(
f"Escrow amount ({amount}) exceeds liability cap ({liability_cap_usd})"
)
perf_escrow_result = api_call("create_performance_escrow", {
"payer_id": self.agent_id,
"payee_id": counterparty_id,
"amount": amount,
"currency": currency,
"description": service_description,
"criteria": sla_metrics,
"metadata": {
"compliance": {
"liability_cap_usd": str(liability_cap_usd),
"timeout_hours": timeout_hours,
"sla_article_15_mapping": {
"accuracy": sla_metrics.get("accuracy_threshold"),
"robustness": sla_metrics.get("uptime_percent"),
},
"applicable_regulations": [
"eu_ai_act_2024_1689",
"product_liability_directive_2024_2853",
],
},
},
})
escrow_id = perf_escrow_result.get("result", {}).get("escrow_id", "unknown")
# Audit trail
self.audit.record_escrow_lifecycle(
escrow_id=escrow_id,
event="created",
amount=amount,
counterparty_id=counterparty_id,
terms={
"type": "performance_escrow",
"sla_metrics": sla_metrics,
"liability_cap_usd": str(liability_cap_usd),
"timeout_hours": timeout_hours,
},
)
return perf_escrow_result
def release_with_compliance_check(
self,
escrow_id: str,
counterparty_id: str,
amount: float,
compliance_verified: bool,
verification_details: str,
) -> dict:
"""
Release escrow only after confirming compliance requirements are met.
Records the release decision and compliance status in the audit chain.
"""
if not compliance_verified:
# Record the failed compliance check
self.audit.record_event(
event_type=ComplianceEventType.COMPLIANCE_CHECK_FAILED,
details={
"escrow_id": escrow_id,
"reason": verification_details,
"action": "release_blocked",
},
counterparty_id=counterparty_id,
financial_amount=amount,
risk_flags=["compliance_block"],
)
raise ValueError(
f"Compliance check failed for escrow {escrow_id}: "
f"{verification_details}"
)
# Record successful compliance check
self.audit.record_event(
event_type=ComplianceEventType.COMPLIANCE_CHECK_PASSED,
details={
"escrow_id": escrow_id,
"verification_details": verification_details,
},
counterparty_id=counterparty_id,
financial_amount=amount,
)
# Release the escrow
# (Actual release mechanism depends on escrow type)
self.audit.record_escrow_lifecycle(
escrow_id=escrow_id,
event="released",
amount=amount,
counterparty_id=counterparty_id,
)
return {"escrow_id": escrow_id, "status": "released", "amount": str(amount)}
# --- Usage ---
audit_chain = ComplianceAuditChain(
agent_id="escrow-manager-eu-prod",
chain_namespace="eu_compliance_2026",
)
escrow_mgr = ComplianceEscrow(
agent_id="escrow-manager-eu-prod",
audit_chain=audit_chain,
)
# Create a liability-bounded escrow
escrow = escrow_mgr.create_compliant_escrow(
counterparty_id="data-analyst-agent-v3",
amount=2500.00,
currency="EUR",
service_description="Quarterly financial data analysis and reporting",
liability_cap_usd=5000.00,
timeout_hours=72,
)
# Create a performance escrow with SLA
perf_escrow = escrow_mgr.create_performance_escrow(
counterparty_id="ml-inference-agent-v2",
amount=10000.00,
currency="EUR",
service_description="Real-time inference service for product recommendations",
sla_metrics={
"accuracy_threshold": 0.95,
"latency_p99_ms": 200,
"uptime_percent": 99.9,
"error_rate_max": 0.005,
},
liability_cap_usd=25000.00,
timeout_hours=168, # 1 week for performance evaluation
)
Why Liability Caps Matter for the Supply Chain
Under the Product Liability Directive, every entity in the supply chain is jointly liable. Without contractual liability caps between agent deployers, a single defective agent in a multi-agent workflow can expose every participant to unlimited liability. The escrow pattern above does three things to mitigate this:
Caps the financial exposure per transaction to a defined amount that both parties accepted at escrow creation time.
Creates an auditable record of the liability terms (Article 12 compliance).
Provides a mechanism for dispute resolution (Article 9 risk mitigation).
This does not eliminate liability to consumers under the Product Liability Directive -- that liability is strict and cannot be contractually excluded. But it does allocate liability between businesses in the supply chain, which is the primary mechanism for managing financial exposure in multi-agent commerce.
Chapter 6: Agent-to-Agent Service Contracts
Why Machine-Readable Contracts Matter
When Agent A hires Agent B through a marketplace listing, the terms of that engagement are currently defined by the listing description -- a free-text field that no machine can enforce. A compliant agent commerce system needs machine-readable contracts that:
Define liability boundaries enforceable by the escrow system.
Specify data processing terms required by GDPR when personal data flows between agents.
Include dispute resolution clauses that map to the platform's dispute mechanism.
Carry regulatory metadata identifying which regulations apply and which party bears which compliance obligations.
Are auditable -- stored as part of the Article 12 record-keeping trail.
Contract Template Structure
The contract is a JSON document stored as marketplace listing metadata. Both parties reference the same contract by its content hash. The contract is not a legal instrument by itself -- it is a technical specification that your legal counsel translates into binding terms. But it is machine-enforceable, which means your agents can automatically reject contracts that violate compliance policies.
import hashlib
from datetime import datetime, timezone, timedelta
def generate_service_contract(
provider_id: str,
consumer_id: str,
service_name: str,
service_description: str,
pricing: dict,
sla: dict,
liability_terms: dict,
data_processing_terms: dict,
dispute_resolution: dict,
regulatory_scope: Optional[list] = None,
contract_duration_days: int = 90,
) -> dict:
"""
Generate a machine-readable service contract for agent-to-agent
commerce. The contract is stored as marketplace listing metadata
and referenced by its content hash.
Args:
provider_id: Agent ID of the service provider.
consumer_id: Agent ID of the consumer (or "*" for open listing).
service_name: Human-readable service name.
service_description: Detailed description of the service.
pricing: {"model": "per_call"|"flat"|"tiered", "amount": float, "currency": str}
sla: {"uptime": float, "latency_p99_ms": int, "accuracy": float}
liability_terms: {
"provider_liability_cap_usd": float,
"consumer_liability_cap_usd": float,
"indemnification": str,
"force_majeure": bool,
}
data_processing_terms: {
"personal_data_processed": bool,
"data_retention_days": int,
"data_location": str,
"gdpr_role": "controller"|"processor"|"none",
"dpa_reference": str | None,
}
dispute_resolution: {
"mechanism": "platform"|"arbitration"|"court",
"escalation_hours": int,
"governing_law": str,
}
regulatory_scope: List of applicable regulation identifiers.
contract_duration_days: How long the contract is valid.
Returns:
The contract document with its content hash.
"""
if regulatory_scope is None:
regulatory_scope = [
"eu_ai_act_2024_1689",
"product_liability_directive_2024_2853",
"gdpr_2016_679",
]
now = datetime.now(timezone.utc)
contract = {
"schema_version": "1.0.0",
"contract_type": "agent_service_agreement",
"parties": {
"provider": provider_id,
"consumer": consumer_id,
},
"service": {
"name": service_name,
"description": service_description,
},
"pricing": pricing,
"sla": sla,
"liability": liability_terms,
"data_processing": data_processing_terms,
"dispute_resolution": dispute_resolution,
"regulatory": {
"applicable_regulations": regulatory_scope,
"provider_compliance_obligations": [
"article_12_record_keeping",
"article_13_transparency",
"article_15_accuracy_robustness",
],
"consumer_compliance_obligations": [
"article_14_human_oversight",
"article_26_deployer_obligations",
],
},
"temporal": {
"created_at": now.isoformat(),
"effective_from": now.isoformat(),
"expires_at": (now + timedelta(days=contract_duration_days)).isoformat(),
"duration_days": contract_duration_days,
},
}
# Compute content hash for tamper detection and referencing
contract_hash = hashlib.sha256(
json.dumps(contract, sort_keys=True).encode()
).hexdigest()
contract["contract_hash"] = contract_hash
return contract
# --- Usage ---
contract = generate_service_contract(
provider_id="ml-inference-agent-v2",
consumer_id="escrow-manager-eu-prod",
service_name="Real-Time Product Recommendation Inference",
service_description=(
"Provides real-time ML inference for product recommendations. "
"Input: user context vector (256-dim). Output: ranked product IDs "
"with confidence scores. Model: transformer-based collaborative "
"filtering, retrained weekly."
),
pricing={
"model": "per_call",
"amount": 0.002,
"currency": "EUR",
},
sla={
"uptime_percent": 99.9,
"latency_p99_ms": 200,
"accuracy_threshold": 0.95,
"error_rate_max": 0.005,
},
liability_terms={
"provider_liability_cap_usd": 25000,
"consumer_liability_cap_usd": 10000,
"indemnification": (
"Provider indemnifies consumer against claims arising from "
"provider's breach of SLA or data processing terms."
),
"force_majeure": True,
},
data_processing_terms={
"personal_data_processed": True,
"data_retention_days": 30,
"data_location": "EU",
"gdpr_role": "processor",
"dpa_reference": "DPA-2026-ACME-ML-001",
},
dispute_resolution={
"mechanism": "platform",
"escalation_hours": 48,
"governing_law": "German law",
},
)
print(f"Contract hash: {contract['contract_hash']}")
Registering Services with Compliance Tags
When listing a service on the marketplace, attach the contract as metadata and use compliance tags to enable discovery by compliance-aware agents.
def register_compliant_service(
agent_id: str,
service_name: str,
description: str,
contract: dict,
risk_level: str,
audit_chain: ComplianceAuditChain,
) -> dict:
"""
Register a marketplace service with compliance tags and contract terms.
Enables other agents to discover services that meet their compliance
requirements.
"""
tags = [
"eu_ai_act_compliant",
f"risk_level:{risk_level}",
"article_12_audited",
"liability_bounded",
f"governing_law:{contract['dispute_resolution']['governing_law']}",
]
if contract["data_processing"]["personal_data_processed"]:
tags.append("gdpr_dpa_required")
tags.append(f"data_location:{contract['data_processing']['data_location']}")
result = api_call("register_service", {
"agent_id": agent_id,
"name": service_name,
"description": description,
"tags": tags,
"metadata": {
"contract_hash": contract["contract_hash"],
"contract": contract,
"compliance_version": "1.0.0",
},
})
# Record in audit trail
audit_chain.record_event(
event_type=ComplianceEventType.AGENT_REGISTERED,
details={
"service_name": service_name,
"contract_hash": contract["contract_hash"],
"risk_level": risk_level,
"tags": tags,
},
)
return result
# --- Contract validation ---
def validate_counterparty_contract(contract: dict, requirements: dict) -> dict:
"""
Validate that a counterparty's contract meets your compliance
requirements before entering into a transaction.
Args:
contract: The counterparty's service contract.
requirements: Your minimum requirements, e.g.:
{
"max_provider_liability_cap": 50000,
"min_uptime": 99.0,
"required_data_location": "EU",
"required_governing_law": "German law",
"gdpr_dpa_required": True,
}
"""
violations = []
# Check liability cap
cap = contract.get("liability", {}).get("provider_liability_cap_usd", 0)
if cap < requirements.get("min_provider_liability_cap", 0):
violations.append(
f"Provider liability cap ({cap}) below minimum "
f"({requirements['min_provider_liability_cap']})"
)
# Check SLA
uptime = contract.get("sla", {}).get("uptime_percent", 0)
if uptime < requirements.get("min_uptime", 0):
violations.append(
f"SLA uptime ({uptime}%) below minimum ({requirements['min_uptime']}%)"
)
# Check data location
data_loc = contract.get("data_processing", {}).get("data_location", "")
required_loc = requirements.get("required_data_location")
if required_loc and data_loc != required_loc:
violations.append(
f"Data location ({data_loc}) does not match requirement ({required_loc})"
)
# Check GDPR DPA
if requirements.get("gdpr_dpa_required"):
dpa_ref = contract.get("data_processing", {}).get("dpa_reference")
if not dpa_ref:
violations.append("GDPR Data Processing Agreement reference missing")
# Check governing law
required_law = requirements.get("required_governing_law")
gov_law = contract.get("dispute_resolution", {}).get("governing_law", "")
if required_law and gov_law != required_law:
violations.append(
f"Governing law ({gov_law}) does not match requirement ({required_law})"
)
return {
"valid": len(violations) == 0,
"violations": violations,
"contract_hash": contract.get("contract_hash"),
"checked_at": datetime.now(timezone.utc).isoformat(),
}
# Validate before transacting
validation = validate_counterparty_contract(
contract=contract,
requirements={
"min_provider_liability_cap": 20000,
"min_uptime": 99.0,
"required_data_location": "EU",
"required_governing_law": "German law",
"gdpr_dpa_required": True,
},
)
if not validation["valid"]:
print(f"Contract rejected: {validation['violations']}")
else:
print("Contract meets all compliance requirements -- proceed with escrow")
Required Contract Clauses Checklist
Every agent-to-agent service contract should include these clauses. Missing any of them creates regulatory or liability exposure:
Clause
Regulatory Basis
Purpose
Liability cap
Product Liability Directive
Bounds financial exposure per party
SLA definition
Article 15 (accuracy, robustness)
Contractually defines "adequate" performance
Dispute resolution
Article 9 (risk management)
Provides fallback when automated systems fail
Data processing terms
GDPR Article 28
Required when personal data flows between agents
Compliance obligations split
Articles 16, 26
Allocates provider vs. deployer obligations
Record-keeping commitment
Article 12
Both parties agree to maintain audit trails
Human oversight trigger
Article 14
Defines when human intervention is required
Termination conditions
Product Liability Directive
Allows exit if counterparty compliance lapses
Force majeure
General contract law
Protects against liability for events beyond control
Governing law
Brussels I Regulation
Determines which jurisdiction resolves disputes
Chapter 7: Compliance Monitoring & Reporting
Article 9: Risk Management System
Article 9 requires a risk management system that is a "continuous iterative process planned and run throughout the entire lifecycle" of the AI system. This is not a one-time risk assessment -- it is a continuous monitoring system that:
Identifies and analyzes known and reasonably foreseeable risks.
Estimates and evaluates risks that may emerge during use.
Evaluates other risks based on post-market monitoring data.
Adopts appropriate risk mitigation measures.
For agent commerce, "continuous monitoring" means automated compliance metric submission, real-time anomaly detection, and webhook-based incident notification. The GreenHelix event and metrics tools provide the infrastructure.
Automated Compliance Metric Submission
import time
from datetime import datetime, timezone
class ComplianceMonitor:
"""
Continuous compliance monitoring for agent commerce systems.
Implements Article 9 risk management through automated metric
submission and threshold-based alerting.
"""
def __init__(
self,
agent_id: str,
audit_chain: ComplianceAuditChain,
thresholds: Optional[dict] = None,
):
self.agent_id = agent_id
self.audit = audit_chain
self.thresholds = thresholds or {
"error_rate_max": 0.01,
"latency_p99_max_ms": 500,
"uptime_min_percent": 99.0,
"dispute_rate_max": 0.05,
"compliance_score_min": 0.80,
}
self.incident_webhooks = []
def register_incident_webhook(self, webhook_url: str, events: list) -> dict:
"""
Register a webhook for compliance incident notifications.
This enables Article 14 human oversight by routing alerts
to human operators when thresholds are breached.
"""
result = api_call("register_webhook", {
"agent_id": self.agent_id,
"url": webhook_url,
"events": events,
})
self.incident_webhooks.append({
"url": webhook_url,
"events": events,
})
return result
def submit_compliance_metrics(self, metrics: dict) -> dict:
"""
Submit compliance metrics to GreenHelix. These form the
quantitative basis for Article 9 risk evaluation.
Expected metrics:
{
"error_rate": float, # Current error rate
"latency_p99_ms": int, # P99 latency
"uptime_percent": float, # Rolling uptime percentage
"transactions_total": int, # Total transactions processed
"disputes_opened": int, # Disputes in current period
"disputes_resolved": int, # Disputes resolved in current period
"escrows_active": int, # Currently active escrows
"compliance_checks_passed": int,
"compliance_checks_failed": int,
}
"""
# Add compliance metadata
metrics["submission_timestamp"] = datetime.now(timezone.utc).isoformat()
metrics["monitor_version"] = "1.0.0"
# Calculate derived compliance score
total_checks = (
metrics.get("compliance_checks_passed", 0)
+ metrics.get("compliance_checks_failed", 0)
)
if total_checks > 0:
metrics["compliance_score"] = (
metrics["compliance_checks_passed"] / total_checks
)
else:
metrics["compliance_score"] = 1.0
# Submit to GreenHelix
result = api_call("submit_metrics", {
"agent_id": self.agent_id,
"metrics": metrics,
})
# Check thresholds and raise incidents
violations = self._check_thresholds(metrics)
if violations:
self._raise_compliance_incident(violations, metrics)
return result
def _check_thresholds(self, metrics: dict) -> list:
"""Check submitted metrics against compliance thresholds."""
violations = []
if metrics.get("error_rate", 0) > self.thresholds["error_rate_max"]:
violations.append({
"metric": "error_rate",
"value": metrics["error_rate"],
"threshold": self.thresholds["error_rate_max"],
"severity": "high",
})
if metrics.get("latency_p99_ms", 0) > self.thresholds["latency_p99_max_ms"]:
violations.append({
"metric": "latency_p99_ms",
"value": metrics["latency_p99_ms"],
"threshold": self.thresholds["latency_p99_max_ms"],
"severity": "medium",
})
uptime = metrics.get("uptime_percent", 100)
if uptime < self.thresholds["uptime_min_percent"]:
violations.append({
"metric": "uptime_percent",
"value": uptime,
"threshold": self.thresholds["uptime_min_percent"],
"severity": "high",
})
# Dispute rate calculation
total_tx = metrics.get("transactions_total", 0)
disputes = metrics.get("disputes_opened", 0)
if total_tx > 0:
dispute_rate = disputes / total_tx
if dispute_rate > self.thresholds["dispute_rate_max"]:
violations.append({
"metric": "dispute_rate",
"value": dispute_rate,
"threshold": self.thresholds["dispute_rate_max"],
"severity": "high",
})
compliance_score = metrics.get("compliance_score", 1.0)
if compliance_score < self.thresholds["compliance_score_min"]:
violations.append({
"metric": "compliance_score",
"value": compliance_score,
"threshold": self.thresholds["compliance_score_min"],
"severity": "critical",
})
return violations
def _raise_compliance_incident(self, violations: list, metrics: dict) -> None:
"""
Raise a compliance incident when thresholds are breached.
Publishes to event bus and records in audit chain.
"""
incident = {
"incident_type": "compliance_threshold_breach",
"agent_id": self.agent_id,
"timestamp": datetime.now(timezone.utc).isoformat(),
"violations": violations,
"metrics_snapshot": metrics,
"severity": max(v["severity"] for v in violations),
}
# Publish incident event
api_call("publish_event", {
"agent_id": self.agent_id,
"event_type": "compliance_incident",
"payload": incident,
})
# Record in audit chain
self.audit.record_event(
event_type=ComplianceEventType.INCIDENT_REPORTED,
details=incident,
risk_flags=[v["metric"] for v in violations],
)
def get_compliance_dashboard(self) -> dict:
"""
Retrieve a compliance dashboard with current metrics,
billing summary, and usage analytics.
"""
# Get usage analytics for cost transparency (Article 13)
usage = api_call("get_usage_analytics", {
"agent_id": self.agent_id,
})
# Get billing summary for financial transparency
billing = api_call("get_billing_summary", {
"agent_id": self.agent_id,
})
# Search for agents with similar compliance profiles
peers = api_call("search_agents_by_metrics", {
"query": {
"capabilities": ["eu_ai_act_compliant"],
},
})
return {
"agent_id": self.agent_id,
"usage_analytics": usage.get("result"),
"billing_summary": billing.get("result"),
"compliant_peers": peers.get("result"),
"dashboard_generated_at": datetime.now(timezone.utc).isoformat(),
}
def run_periodic_check(self, metrics_source: callable, interval_seconds: int = 300):
"""
Run continuous compliance monitoring. Call this in a background
thread or scheduled task.
Args:
metrics_source: Callable that returns current metrics dict.
interval_seconds: How often to check (default 5 minutes).
"""
while True:
try:
current_metrics = metrics_source()
self.submit_compliance_metrics(current_metrics)
except Exception as e:
# Record monitoring failures as incidents
self.audit.record_event(
event_type=ComplianceEventType.INCIDENT_REPORTED,
details={
"error": str(e),
"monitoring_failure": True,
},
risk_flags=["monitoring_gap"],
)
time.sleep(interval_seconds)
# --- Usage ---
audit_chain = ComplianceAuditChain(
agent_id="escrow-manager-eu-prod",
chain_namespace="eu_compliance_2026",
)
monitor = ComplianceMonitor(
agent_id="escrow-manager-eu-prod",
audit_chain=audit_chain,
thresholds={
"error_rate_max": 0.01,
"latency_p99_max_ms": 300,
"uptime_min_percent": 99.9,
"dispute_rate_max": 0.02,
"compliance_score_min": 0.90,
},
)
# Register webhooks for human oversight (Article 14)
monitor.register_incident_webhook(
webhook_url="https://ops.acme.example/compliance-alerts",
events=["compliance_incident", "dispute_opened", "escrow_timeout"],
)
# Submit current metrics
monitor.submit_compliance_metrics({
"error_rate": 0.003,
"latency_p99_ms": 180,
"uptime_percent": 99.97,
"transactions_total": 14200,
"disputes_opened": 2,
"disputes_resolved": 2,
"escrows_active": 47,
"compliance_checks_passed": 283,
"compliance_checks_failed": 1,
})
# Get compliance dashboard
dashboard = monitor.get_compliance_dashboard()
print(json.dumps(dashboard, indent=2))
Webhook-Based Incident Notification for Article 14
Article 14 requires that high-risk AI systems can be "effectively overseen by natural persons." In agent commerce, where transactions happen at machine speed, effective oversight requires automated alerting that routes incidents to human operators in real time. The webhook pattern above achieves this:
Threshold breach alerts: When compliance metrics cross defined thresholds, the system publishes an event that triggers webhook delivery to the operations team.
Dispute notifications: When a counterparty opens a dispute, the webhook fires immediately, giving the human operator time to intervene before the dispute resolution timeout.
Escrow timeout warnings: When an escrow approaches its timeout, the webhook alerts the operator to either release, extend, or cancel before the automatic timeout triggers.
The key design principle is that the system never requires a human to be watching continuously. It requires a human to be reachable within the defined escalation window. This is the practical interpretation of Article 14 for autonomous agent systems.
Dispute Resolution for Contractual Fallback
When automated compliance monitoring detects a breach, and the counterparty disagrees with the assessment, the dispute resolution mechanism provides the contractual fallback:
def open_compliance_dispute(
agent_id: str,
counterparty_id: str,
escrow_id: str,
reason: str,
evidence: dict,
audit_chain: ComplianceAuditChain,
) -> dict:
"""
Open a formal dispute with compliance-relevant evidence.
Records the dispute in the audit chain for Article 12.
"""
dispute_result = api_call("open_dispute", {
"agent_id": agent_id,
"counterparty_id": counterparty_id,
"escrow_id": escrow_id,
"reason": reason,
"evidence": evidence,
})
audit_chain.record_event(
event_type=ComplianceEventType.DISPUTE_OPENED,
details={
"escrow_id": escrow_id,
"reason": reason,
"evidence_summary": list(evidence.keys()),
},
counterparty_id=counterparty_id,
)
return dispute_result
def resolve_compliance_dispute(
agent_id: str,
dispute_id: str,
resolution: str,
counterparty_id: str,
audit_chain: ComplianceAuditChain,
) -> dict:
"""
Resolve a dispute and record the resolution in the audit trail.
"""
resolve_result = api_call("resolve_dispute", {
"agent_id": agent_id,
"dispute_id": dispute_id,
"resolution": resolution,
})
audit_chain.record_event(
event_type=ComplianceEventType.DISPUTE_RESOLVED,
details={
"dispute_id": dispute_id,
"resolution": resolution,
},
counterparty_id=counterparty_id,
)
return resolve_result
Chapter 8: The Pre-August 2026 Compliance Sprint
12-Week Sprint to Compliance
August 2, 2026 is not negotiable. If you are reading this guide in April 2026, you have approximately 16 weeks. Here is a 12-week sprint plan that leaves 4 weeks of buffer for testing, legal review, and unexpected regulatory guidance. Each week has specific deliverables tied to the code patterns in this guide.
Weeks 1-2: Foundation
Goal: Risk classification complete, agent identities registered with compliance metadata.
Run classify_agent_risk() (Chapter 2) against every agent in production.
Document classification decisions and reasoning. Store in version control.
Register all agents using ComplianceAgent.register_with_compliance() (Chapter 3).
Engage legal counsel to validate risk classifications. Flag any edge cases from the "elevated" risk category.
Create a compliance metadata schema that your legal team approves.
Quick win: Risk classification can be done in a single afternoon. The code in Chapter 2 produces a structured output that your legal team can review.
Weeks 3-4: Audit Infrastructure
Goal: Cryptographic audit trails operational for all compliance-relevant events.
Deploy ComplianceAuditChain (Chapter 4) for every production agent.
Integrate audit chain recording into all escrow create/release/cancel paths.
Integrate audit chain recording into all dispute open/resolve paths.
Set up verify_audit_completeness() as a daily cron job.
Verify that chain integrity checks pass for all agents.
Quick win: If you already use GreenHelix for escrow and payments, adding the audit chain calls is wrapping existing code with three additional lines per operation.
Weeks 5-6: Contracts & Liability
Goal: Machine-readable service contracts for all marketplace listings.
Generate service contracts using generate_service_contract() (Chapter 6) for every active marketplace listing.
Have legal counsel review the contract template and approve the required clauses.
Implement validate_counterparty_contract() in your agent's decision loop -- no transaction without a valid contract.
Update all escrows to use ComplianceEscrow.create_compliant_escrow() (Chapter 5) with liability caps.
Backfill compliance metadata on existing active escrows.
Deep work: The contract template requires legal review. Start this process in Week 3 so the template is approved by Week 5.
Weeks 7-8: Monitoring & Alerting
Goal: Continuous compliance monitoring with human oversight webhooks.
Deploy ComplianceMonitor (Chapter 7) for every production agent.
Configure compliance thresholds based on your SLA commitments.
Register incident webhooks routed to your operations team's alerting system (PagerDuty, Opsgenie, etc.).
Test the full alerting pipeline: inject a threshold breach and verify the webhook fires, the event publishes, and the audit chain records the incident.
Set up the compliance dashboard and schedule weekly review meetings with your legal/compliance team.
