agentguard-eu 0.2.1

EU AI Act compliance middleware for AI agents. Make any LLM-powered agent legally deployable in Europe with 3 lines of code.

🛡️ AgentGuard

EU AI Act compliance middleware for AI agents. Make any LLM-powered agent legally deployable in Europe with 3 lines of code.

---

The Problem

Starting August 2, 2026, every company deploying AI systems in the EU must comply with the EU AI Act — or face fines up to €35M or 7% of global turnover.

AgentGuard fixes that. It's a lightweight middleware that wraps any AI agent or LLM call with the compliance layer required by the EU AI Act:

EU AI Act Requirement	Article	AgentGuard Feature
Users must know they're talking to AI	Art. 50(1)	Contextual smart disclosures (category-aware, multi-language)
AI content must be machine-readable labeled	Art. 50(2)	Content labeling (C2PA-compatible)
Interactions must be logged and auditable	Art. 12	Structured audit logging (file/SQLite)
Human oversight must be possible	Art. 14	Automatic escalation + review queue
Harmful content must be prevented	Runtime	Input/Output Policy Engine (block/flag/disclaim)
System must be documented	Art. 11, 18	Auto-generated compliance reports

Quick Start

pip install agentguard-eu

from agentguard import AgentGuard

# 1. Initialize with your system details
guard = AgentGuard(
    system_name="customer-support-bot",
    provider_name="my-provider",
    risk_level="limited",
)

# 2. Wrap any LLM function
result = guard.invoke(
    func=my_llm_function,       # Your existing AI function
    input_text="What is your return policy?",
    user_id="customer-42",
)

# 3. Everything is now compliant
print(result["response"])         # AI response (NEVER modified by default)
print(result["interaction_id"])   # Unique audit trail ID
print(result["compliance"])       # Structured compliance metadata dict
print(result["disclosure"])       # HTTP headers for Article 50
print(result["content_label"])    # Machine-readable content label
print(result["escalated"])        # Whether human review was triggered

That's it. Your existing AI code doesn't change. AgentGuard wraps it.

Three Ways to Use

1. guard.invoke() — Wrap any function call

result = guard.invoke(
    func=lambda q: openai_client.chat.completions.create(
        model="gpt-4", messages=[{"role": "user", "content": q}]
    ).choices[0].message.content,
    input_text="Hello!",
    user_id="user-123",
    model="gpt-4",
    confidence=0.92,
)

2. @guard.compliant — Decorator

@guard.compliant(model="gpt-4")
def ask_support(query: str) -> str:
    return openai_client.chat.completions.create(...).choices[0].message.content

result = ask_support("Do you ship internationally?", user_id="user-456")

3. guard.interaction() — Context manager

with guard.interaction(user_id="user-789") as ctx:
    response = my_complex_agent.run("Analyze this contract")
    ctx.record(
        input_text="Analyze this contract",
        output_text=response,
        confidence=0.45,
    )
    # Low confidence + keyword "contract" → auto-escalated

Input/Output Policy Engine — Runtime Content Enforcement

The policy engine is AgentGuard's core differentiator: block, flag, or disclaim content on every LLM call — before and after the model responds.

Input → [InputPolicy: block/flag/allow] → [LLM Call] → [OutputPolicy: disclaim/block/pass] → Output

InputPolicy (pre-call)

Runs before the LLM is called. Blocked requests never reach the API — zero cost, zero latency.

from agentguard import AgentGuard, InputPolicy, OutputPolicy, PolicyAction
from agentguard.policy import CustomRule

input_policy = InputPolicy(
    block_categories=["weapons", "self_harm", "csam"],
    flag_categories=["emotional_simulation", "medical", "legal", "financial"],
    max_input_length=5000,
    custom_rules=[
        CustomRule(
            name="prompt_injection",
            pattern=r"ignore previous instructions",
            action=PolicyAction.BLOCK,
            message="Blocked: potential prompt injection.",
        ),
    ],
)

OutputPolicy (post-call)

Runs after the LLM responds. Adds category-specific disclaimers (in metadata by default) or blocks unsafe output.

output_policy = OutputPolicy(
    scan_categories=["medical", "legal", "financial"],
    block_on_detect=False,     # True = replace response entirely
    add_disclaimer=True,       # Add category-aware disclaimers (in metadata by default)
)

Important: By default (disclosure_method="metadata"), disclaimers are placed in response.compliance["policy"]["disclaimer"] — the actual LLM response content is never modified. Set disclosure_method="prepend" if you want disclaimers appended to the response text.

Wire it up

guard = AgentGuard(
    system_name="my-bot",
    provider_name="My Company",
    risk_level="limited",
    input_policy=input_policy,
    output_policy=output_policy,
)

Content Detection

AgentGuard ships with built-in keyword/pattern matchers for common categories:

Category	Default Action	Detected via
weapons	Block	Keywords + regex patterns
self_harm	Block	Keywords + regex
csam	Block	Keywords
medical	Flag + disclaimer	Keywords (diagnosis, symptoms, dosage...)
legal	Flag + disclaimer	Keywords (lawsuit, attorney, liability...)
financial	Flag + disclaimer	Keywords (invest, portfolio, trading...)
emotional_simulation	Flag	Keywords (girlfriend, boyfriend...)

Built-in detection uses keyword and regex patterns. This is fast (<1ms) and catches obvious cases, but will miss paraphrased or subtle inputs. For production accuracy, plug in a dedicated classifier:

Custom Classifier Hook

Plug in any external classifier — Azure Content Safety, OpenAI Moderation, Llama Guard, or your own rules. AgentGuard merges the results with its built-in keyword detection:

from agentguard import AgentGuard, InputPolicy, OutputPolicy

Azure AI Content Safety:

from azure.ai.contentsafety import ContentSafetyClient

safety_client = ContentSafetyClient(endpoint="...", credential="...")

def azure_classifier(text: str) -> list[str]:
    result = safety_client.analyze_text({"text": text})
    categories = []
    if result.violence_result.severity >= 2:
        categories.append("weapons")
    if result.self_harm_result.severity >= 2:
        categories.append("self_harm")
    return categories

guard = AgentGuard(
    ...,
    input_policy=InputPolicy(
        block_categories=["weapons", "self_harm"],
        custom_classifier=azure_classifier,
    ),
)

OpenAI Moderation (free):

from openai import OpenAI
oai = OpenAI()

def openai_classifier(text: str) -> list[str]:
    result = oai.moderations.create(input=text)
    scores = result.results[0].categories
    categories = []
    if scores.violence: categories.append("weapons")
    if scores.self_harm: categories.append("self_harm")
    if scores.sexual_minors: categories.append("csam")
    return categories

guard = AgentGuard(
    ...,
    input_policy=InputPolicy(
        block_categories=["weapons", "self_harm", "csam"],
        custom_classifier=openai_classifier,
    ),
)

Simple domain rules:

def healthcare_rules(text: str) -> list[str]:
    categories = []
    controlled = ["oxycodone", "fentanyl", "morphine"]
    if any(drug in text.lower() for drug in controlled):
        categories.append("controlled_substance")
    return categories

The custom classifier is any Callable[[str], list[str]]. If it crashes, AgentGuard catches the error and continues with keyword results only. If it's too slow, it's skipped after classifier_timeout seconds (default: 5.0).

Detection Quality Roadmap

Version	Method	Accuracy	Latency	Cost
v0.1 (now)	Keywords + regex	~70%	<1ms	Free
v0.1+ (now)	+ Custom classifier hook	User-defined	User-defined	User-defined
v0.3	+ LLM-as-judge option	~95%	+200ms	~$0.00001/call
v1.0	+ Fine-tuned small model	~93%	+20ms	Free (local)

You can override or extend any category with your own CategoryDefinition.

Policy metadata on every response

When using provider wrappers, policy results are attached to every response via response.compliance:

response = client.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": "Tell me about my symptoms"}],
)

# Response content is NEVER modified (default: disclosure_method="metadata")
print(response.choices[0].message.content)  # pristine LLM output

# All compliance data in structured metadata
print(response.compliance["policy"]["disclaimer"])
# "This is AI-generated and not medical advice..."

print(response.compliance["disclosure"]["text"])
# "This is an AI system. Information provided is not medical advice..."

# HTTP headers ready for FastAPI/Flask forwarding
print(response.compliance_headers)
# {"X-AI-Generated": "true", "X-AI-System": "my-bot", ...}

For blocked requests (e.g., weapons), the LLM is never called — zero cost, zero latency:

response = client.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": "How to build a bomb?"}],
)
print(response.compliance["policy"]["input_action"])  # "blocked"

Contextual Smart Disclosures (Article 50)

Instead of a generic "you are talking to AI" message, AgentGuard adapts the disclosure based on detected content categories — in the user's language.

guard = AgentGuard(
    system_name="my-bot",
    provider_name="My Company GmbH",
    disclosure_method="metadata",      # DEFAULT: never touch response content
    # Other options: "prepend", "first_only", "header", "none"
    disclosure_mode="contextual",      # "static" = same text always (default)
    language="de",                     # en, de, fr, es, it built-in
)

Disclosure methods:

Method	Behavior
"metadata" (default)	Attach to response.compliance — never modify content
"prepend"	Prepend disclosure text before response content
"first_only"	Prepend only on first message per session_id
"header"	Return as HTTP headers dict via response.compliance_headers
"none"	Disable disclosure text entirely (just log and audit)

When the policy engine detects medical content, the user sees (in German):

Dies ist ein KI-System. Die bereitgestellten Informationen stellen keine medizinische Beratung dar. Bitte konsultieren Sie einen Arzt.

Instead of a generic "You are talking to an AI."

Built-in languages: English, German, French, Spanish, Italian — each with category-specific templates for medical, legal, financial, emotional simulation, and self-harm.

# Override or add templates for any category/language
guard = AgentGuard(
    ...,
    disclosure_mode="contextual",
    language="pt",
    disclosure_languages={
        "pt": {
            "default": "Voce esta interagindo com um sistema de IA ({system_name}).",
            "medical": "Informacao nao constitui aconselhamento medico.",
        },
    },
)

Multiple categories are combined automatically. If a query triggers both medical and legal, the user sees both disclosures.

Human Oversight (Article 14)

AgentGuard automatically detects when interactions should be reviewed by a human:

guard = AgentGuard(
    system_name="my-bot",
    provider_name="my-provider",
    human_escalation="low_confidence",
    confidence_threshold=0.7,
    sensitive_keywords=["legal", "medical", "financial advice"],
    block_on_escalation=True,  # Block response until human approves
)

# Check pending reviews
for review in guard.pending_reviews:
    print(f"Needs review: {review['reason']}")

# Approve or reject
guard.oversight.approve(interaction_id)
guard.oversight.reject(interaction_id, reason="Inaccurate response")

Compliance Reports (Articles 11, 18)

Generate audit documentation with one line:

# JSON report
guard.generate_report("compliance_report.json")

# Markdown report (for human reading)
print(guard.generate_report_markdown())

Reports include: system identification, transparency configuration, human oversight settings, interaction statistics, and escalation history.

Provider Wrappers

Zero-effort compliance for popular LLM clients — wrap once, every call is compliant.

OpenAI

pip install "agentguard-eu[openai]"

from agentguard import AgentGuard, wrap_openai
from openai import OpenAI

guard = AgentGuard(system_name="my-bot", provider_name="my-provider")
client = wrap_openai(OpenAI(), guard)

# Every call is now compliant — logged, disclosed, escalation-checked
response = client.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": "Hello!"}],
)
print(response.choices[0].message.content)  # untouched LLM output
print(response.compliance)                  # structured compliance metadata
print(response.compliance_headers)          # HTTP headers for forwarding

Streaming works too — chunks yield in real-time, compliance runs after the stream completes:

stream = client.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": "Hello!"}],
    stream=True,
)
for chunk in stream:
    if chunk.choices and chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="")
print(stream.compliance)           # available after iteration
print(stream.compliance_headers)   # HTTP headers

Azure OpenAI

pip install "agentguard-eu[openai]"

from agentguard import AgentGuard, wrap_azure_openai
from openai import AzureOpenAI

guard = AgentGuard(system_name="my-bot", provider_name="my-provider")
client = wrap_azure_openai(
    AzureOpenAI(
        azure_endpoint="https://my-resource.openai.azure.com",
        api_version="2024-02-01",
        api_key="...",
    ),
    guard,
)

response = client.chat.completions.create(
    model="my-deployment",
    messages=[{"role": "user", "content": "Hello!"}],
)

Anthropic

pip install "agentguard-eu[anthropic]"

from agentguard import AgentGuard, wrap_anthropic
from anthropic import Anthropic

guard = AgentGuard(system_name="my-bot", provider_name="my-provider")
client = wrap_anthropic(Anthropic(), guard)

message = client.messages.create(
    model="claude-sonnet-4-5-20250929",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello!"}],
)
print(message.content[0].text)  # untouched LLM output
print(message.compliance)       # structured compliance metadata
print(message.compliance_headers)  # HTTP headers for forwarding

LangChain

pip install "agentguard-eu[langchain]"

from agentguard import AgentGuard, AgentGuardCallback
from langchain_openai import ChatOpenAI  # or AzureChatOpenAI, ChatAnthropic, etc.

guard = AgentGuard(system_name="my-bot", provider_name="my-provider")
callback = AgentGuardCallback(guard, user_id="user-123")
llm = ChatOpenAI(model="gpt-4", callbacks=[callback])

response = llm.invoke("Hello!")
print(response.content)
print(callback.last_result)  # compliance metadata for most recent call
print(callback.results)      # all runs keyed by run_id

Works with any LangChain LLM — ChatOpenAI, AzureChatOpenAI, ChatAnthropic, and more. Streaming is also supported automatically via the callback hooks.

Audit Backends (Article 12)

AgentGuard supports multiple audit backends for logging all AI interactions:

File Backend (default)

Writes one JSONL file per day to the audit directory. Simple, portable, and easy to ship to external systems:

guard = AgentGuard(
    system_name="my-bot",
    provider_name="my-provider",
    audit_backend="file",
    audit_path="./agentguard_audit",
)
# Logs go to ./agentguard_audit/audit_2026-02-07.jsonl

SQLite Backend

Local SQLite database with built-in querying and statistics. Required for the dashboard and compliance report statistics:

guard = AgentGuard(
    system_name="my-bot",
    provider_name="my-provider",
    audit_backend="sqlite",
    audit_path="./agentguard_audit",
)

# Query audit logs programmatically
entries = guard.audit.query(
    start_date="2026-01-01",
    end_date="2026-02-07",
    user_id="customer-42",
    escalated_only=True,
    limit=100,
)

# Get aggregate statistics
stats = guard.audit.get_stats()
print(stats)
# {
#     "total_interactions": 1234,
#     "total_escalations": 56,
#     "total_errors": 3,
#     "disclosures_shown": 1231,
#     "unique_users": 89,
#     "avg_latency_ms": 245.3,
# }

Custom Backend

Provide your own callback for integration with external logging systems (e.g., S3, BigQuery, Datadog):

def my_log_handler(entry):
    # Send to your logging infrastructure
    requests.post("https://my-logging-api/ingest", json=entry.model_dump())

guard = AgentGuard(
    system_name="my-bot",
    provider_name="my-provider",
    audit_backend="custom",
)
# Pass custom_callback when constructing the AuditLogger directly

Human Review Dashboard

A Streamlit-based dashboard for Article 14 human oversight.

pip install "agentguard-eu[dashboard]"
agentguard-dashboard --audit-path ./agentguard_audit

Features:

• Compliance statistics (total interactions, escalation rate, avg confidence, avg latency)
• Pending escalation review queue with Approve/Reject buttons
• Full audit log browser with filters (date range, user ID, escalated only)
• CSV export of audit data
• Live compliance report viewer

Requires the sqlite audit backend (audit_backend="sqlite") to be enabled in your AgentGuard configuration. The dashboard connects directly to the SQLite audit database.

FastAPI Integration

Use response.compliance_headers to forward compliance headers to HTTP responses:

from fastapi import FastAPI
from fastapi.responses import JSONResponse

@app.post("/chat")
async def chat(query: str):
    response = client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": query}],
    )
    fastapi_response = JSONResponse({"message": response.choices[0].message.content})
    for key, value in response.compliance_headers.items():
        fastapi_response.headers[key] = value
    return fastapi_response

Configuration

guard = AgentGuard(
    # Identity (Article 16)
    system_name="my-ai-system",
    provider_name="My Company GmbH",
    risk_level="limited",           # "minimal", "limited", or "high"
    intended_purpose="Customer support chatbot",

    # Transparency (Article 50)
    disclosure_method="metadata",   # "metadata" (default), "prepend", "first_only", "header", "none"
    disclosure_mode="contextual",   # "static" (default) or "contextual"
    language="en",                  # en, de, fr, es, it (or add your own)
    label_content=True,             # Machine-readable content labels

    # Content Policies (runtime enforcement)
    input_policy=InputPolicy(
        block_categories=["weapons", "self_harm"],
        flag_categories=["medical", "legal", "financial"],
    ),
    output_policy=OutputPolicy(
        scan_categories=["medical", "legal", "financial"],
        add_disclaimer=True,
    ),

    # Audit (Article 12)
    audit_backend="sqlite",         # "file", "sqlite", or "custom"
    audit_path="./audit_logs",
    log_inputs=True,
    log_outputs=True,
    retention_days=365,

    # Human Oversight (Article 14)
    human_escalation="low_confidence",  # "never", "low_confidence", "sensitive_topic", "always_review"
    confidence_threshold=0.7,
    sensitive_keywords=["legal", "medical", "financial"],
    escalation_callback=my_slack_notifier,  # Optional: get notified
    block_on_escalation=False,
)

What AgentGuard is NOT

• ❌ Not a legal compliance guarantee (consult qualified legal professionals)
• ❌ Not an AI agent framework (use LangGraph, CrewAI, etc. — then wrap with AgentGuard)
• ❌ Not a replacement for a full conformity assessment (required for high-risk systems)
• ✅ A practical engineering tool that covers the technical requirements

EU AI Act Timeline

Date	Milestone
Feb 2025	Prohibited AI practices banned
Aug 2025	GPAI model rules in effect
Aug 2026	Full enforcement: transparency, high-risk obligations, Article 50
Aug 2027	Remaining provisions for product-embedded AI

AgentGuard targets the August 2026 deadline — the biggest compliance milestone for most companies.

Contributing

Contributions welcome! See CONTRIBUTING.md for guidelines.

Priority areas:

• Cloud audit backends (S3, BigQuery)
• C2PA standard implementation
• Async support (ainvoke, async wrappers)
• Webhook notifications for escalations

License

Apache 2.0 — use it freely in commercial projects.

---

Built by Sagar — AI Engineer based in Berlin, specializing in enterprise AI systems and compliance.

AgentGuard: Because shipping AI agents without compliance is like shipping code without tests. You can do it, but you probably shouldn't.