FIDES: Deterministic Prompt Injection Defense System (Costa et al., 2025)

FIDES is a comprehensive security system for AI agents. This developer guide describes the deterministic prompt injection defense system implemented in the agent framework. The system provides label-based security mechanisms to defend against prompt injection attacks by tracking integrity and confidentiality of content throughout agent execution.

Context Provider Pattern with SecureAgentConfig!

SecureAgentConfig is now a ContextProvider — add it to any agent with a single context_providers=[config] line. It automatically injects security tools, instructions, and middleware via the before_run() hook. No security knowledge required from developers.

Key Features:

Context Provider Pattern - SecureAgentConfig extends ContextProvider, injecting everything automatically
Automatic Variable Hiding - UNTRUSTED content is automatically stored and replaced with references
Per-Item Embedded Labels - Tools return list[Content] with Content.from_text() for proper label propagation
Zero-Config Security - context_providers=[config] replaces manual middleware=, tools=, and instructions= wiring
Variable ID Support - quarantined_llm now accepts variable_ids to directly reference hidden content
Security Instructions - Built-in SECURITY_TOOL_INSTRUCTIONS automatically injected into agent context

Overview

The defense system consists of eight main components:

Content Labeling Infrastructure - Labels for tracking integrity and confidentiality
Label Tracking Middleware - Automatically assigns, propagates labels, and hides untrusted content
Per-Item Embedded Labels - Tools can return mixed-trust data with per-item security labels
Policy Enforcement Middleware - Blocks tool calls that violate security policies
Security Tools - Specialized tools for safe handling of untrusted content (quarantined_llm, inspect_variable)
SecureAgentConfig - Helper class for easy secure agent configuration
Message-Level Label Tracking - Track labels on every message in the conversation (Phase 1)
MCP Auto-Labeling and Result IFC Parsing - Auto-label MCP tools from hints and parse server _meta.ifc labels

Architecture

1. Content Labels

Every piece of content (tool calls, results, messages) can be assigned a ContentLabel with two dimensions:

Integrity Labels

TRUSTED: Content from trusted sources (user input, system messages)
UNTRUSTED: Content from untrusted sources (AI-generated, external APIs)

Confidentiality Labels

PUBLIC: Content can be shared publicly
PRIVATE: Content is private and should not be shared
USER_IDENTITY: Content is restricted to specific user identities only

from agent_framework.security import ConfidentialityLabel, ContentLabel, IntegrityLabel, PRINCIPAL_METADATA_KEY

# Create a label
label = ContentLabel(
    integrity=IntegrityLabel.TRUSTED,
    confidentiality=ConfidentialityLabel.USER_IDENTITY,
    metadata={
        PRINCIPAL_METADATA_KEY: [
            {"tenant_id": "tenant-123", "user_id": "user-123"},
        ]
    },
)

1.1 USER_IDENTITY Principal Binding

USER_IDENTITY labels require a canonical, non-empty principal set. Each principal contains exactly tenant_id and user_id. Build this metadata from the authenticated request or session, or from a locally trusted static tool declaration. Do not infer it from model arguments or remote result metadata.

Source tools declare the owner of identity-scoped output. Destination tools declare the principals they authorize using the same namespaced key:

from agent_framework import tool
from agent_framework.security import PRINCIPAL_METADATA_KEY

alice = [{"tenant_id": "tenant-contoso", "user_id": "alice"}]


@tool(
    description="Read Alice's profile",
    additional_properties={
        "source_integrity": "trusted",
        "confidentiality": "user_identity",
        PRINCIPAL_METADATA_KEY: alice,
    },
)
async def read_profile() -> str:
    return "Alice profile data"


@tool(
    description="Save data to Alice's profile",
    additional_properties={
        "max_allowed_confidentiality": "user_identity",
        PRINCIPAL_METADATA_KEY: alice,
    },
)
async def save_profile(data: str) -> None:
    ...

The policy allows a flow only when every source principal is present in the destination's authorized set. Missing, malformed, or mismatched principal data is blocked. See user_identity_security_example.py for a complete runnable setup.

Migrating legacy USER_IDENTITY labels

Earlier FIDES examples used a single, unnamespaced user_id and did not bind identity destinations. Releases containing principal-bound enforcement reject that shape. Migrate both the source label and every USER_IDENTITY destination; do not fill in a tenant or user from model-generated arguments.

Before:

legacy_label = ContentLabel(
    confidentiality=ConfidentialityLabel.USER_IDENTITY,
    metadata={"user_id": "alice"},
)


@tool(
    description="Save identity data",
    additional_properties={"max_allowed_confidentiality": "user_identity"},
)
async def save_identity_data(data: str) -> None:
    ...

After:

alice = [{"tenant_id": authenticated_tenant_id, "user_id": authenticated_user_id}]

label = ContentLabel(
    confidentiality=ConfidentialityLabel.USER_IDENTITY,
    metadata={PRINCIPAL_METADATA_KEY: alice},
)


@tool(
    description="Save identity data",
    additional_properties={
        "max_allowed_confidentiality": "user_identity",
        PRINCIPAL_METADATA_KEY: alice,
    },
)
async def save_identity_data(data: str) -> None:
    ...

Combined content may contain more than one principal. A destination must list all authorized principals because the policy checks that the source set is a subset of the destination set.

2. Label Tracking Middleware with Tiered Label Propagation

LabelTrackingFunctionMiddleware uses a tiered label propagation scheme where the result label of a tool call is determined by a strict 3-tier priority:

Priority	Source	Used When
Tier 1	Per-item embedded labels (`additional_properties.security_label`)	Restrict the locally established fallback
Tier 2	Tool's `source_integrity` declaration	No embedded labels, but tool declares `source_integrity`
Tier 3 (Lowest)	Owned-reference integrity or configured default, restricted by argument labels	No `source_integrity` declared
Default	`default_integrity` (`UNTRUSTED` by default)	No source declaration or resolved, owned variable references

Tiered Label Propagation:

Tier 1: Embedded labels are restriction-only by default: they can downgrade integrity or raise confidentiality, but cannot upgrade a fallback or supply principal authority
Tier 2: source_integrity is the locally trusted fallback for the tool's output; use "trusted" only after the local connector enforces its trust policy
Tier 3: Owned input baseline — inherit integrity from labels retrieved while resolving variable references owned by the current security scope. If there are no such references, use default_integrity. Labels supplied in arguments may make that baseline less trusted, but cannot make it more trusted.
Default: UNTRUSTED unless the application configures another default_integrity

A security_label or legacy label dictionary in tool arguments is application data, not an authoritative trust declaration, even inside a validated model or nested dictionary. Its integrity claim cannot promote the result above the owned-reference/default baseline. Argument confidentiality restrictions still propagate, while principal authority comes from owned labels or local configuration. An explicit source_integrity declaration retains precedence over argument integrity claims.

Framework-owned parsers and wrappers may stamp a complete label after enforcing local policy. Application and remote metadata remains restriction-only.

Per-Item Embedded Labels (RECOMMENDED for Mixed-Trust Data): Tools returning mixed-trust data should embed labels on each item in additional_properties.security_label:

# Each item has its own security label
[
    {"id": 1, "body": "trusted content", "additional_properties": {"security_label": {"integrity": "trusted"}}},
    {"id": 2, "body": "untrusted content", "additional_properties": {"security_label": {"integrity": "untrusted"}}},
]

The middleware automatically:

Hides items with integrity: "untrusted" → replaced with VariableReferenceContent
Keeps items with integrity: "trusted" visible in LLM context
Combines labels from all items for the overall result label

Tool-Level Source Integrity (Tier 2 Fallback): If items don't have embedded labels, the tool can declare a fallback via source_integrity. When declared, source_integrity establishes the local integrity fallback. Embedded labels can make the result less trusted, but cannot make it more trusted:

source_integrity="trusted": Tool produces trusted data (internal computations)
source_integrity="untrusted": Tool fetches untrusted data
(not set): Falls back to tier 3 (owned-reference integrity or the configured default, restricted by argument labels)

Note: Action tools (sinks like send_email) can omit source_integrity. Their result follows tier 3: owned-reference inheritance or the configured default, with argument restrictions.

Context Label Tracking:

Context label starts as TRUSTED + PUBLIC on first call
Gets updated (tainted) when untrusted content enters the context
Hidden content does NOT taint the context (it never enters LLM context)
Policy enforcement uses the context label for validation

Automatic Hiding:

UNTRUSTED results/items are automatically hidden in variable store
LLM context sees only VariableReferenceContent
Since hidden content doesn't enter context, it doesn't taint the context label

import json
from agent_framework import Content, tool
from agent_framework.security import LabelTrackingFunctionMiddleware, SecureAgentConfig

# Define a tool that returns mixed-trust data with per-item labels
@tool(
    description="Fetch emails from inbox",
    additional_properties={
        "source_integrity": "trusted",  # Local connector verifies internal senders
    },
)
async def fetch_emails(count: int = 5) -> list[Content]:
    """Fetch emails - some from trusted internal sources, others from external sources."""
    emails = get_emails(count)
    return [
        Content.from_text(
            json.dumps({
                "id": email["id"],
                "from": email["from"],
                "subject": email["subject"],
                "body": email["body"],
            }),
            # Per-item label - middleware automatically hides untrusted items
            additional_properties={
                "security_label": {
                    "integrity": "trusted" if email["is_internal"] else "untrusted",
                    "confidentiality": "private",
                }
            },
        )
        for email in emails
    ]

# Define a tool that performs internal (trusted) computation
@tool(
    description="Calculate statistics",
    additional_properties={
        "source_integrity": "trusted",  # Fallback if no per-item labels
    }
)
async def calculate_stats(data: dict) -> dict:
    # Because source_integrity is declared, output trust comes from the tool
    # declaration (tier 2), not from argument label joins.
    return {"mean": 42}

# Recommended: Use SecureAgentConfig as a context provider
config = SecureAgentConfig(
    auto_hide_untrusted=True,
    allow_untrusted_tools={"fetch_emails"},
    block_on_violation=True,
)

agent = Agent(
    client=client,
    name="assistant",
    instructions="You are a helpful assistant.",
    tools=[fetch_emails, calculate_stats],
    context_providers=[config],  # Injects tools, instructions, and middleware automatically
)

3. Per-Item Embedded Labels

For tools that return mixed-trust data (e.g., emails from both internal and external sources), you can embed security labels on individual items using additional_properties.security_label:

import json
from agent_framework import Content, tool

@tool(
    description="Fetch emails from inbox",
    additional_properties={
        "source_integrity": "trusted",  # Local connector verifies internal senders
    },
)
async def fetch_emails(count: int = 5) -> list[Content]:
    """Fetch emails with per-item security labels."""
    emails = fetch_from_server(count)

    return [
        Content.from_text(
            json.dumps({
                "id": email["id"],
                "from": email["from"],
                "subject": email["subject"],
                "body": email["body"],
            }),
            # Embed security label for this specific item
            additional_properties={
                "security_label": {
                    "integrity": "trusted" if is_internal_sender(email["from"]) else "untrusted",
                    "confidentiality": "private",
                }
            },
        )
        for email in emails
    ]

How It Works:

Tool returns mixed-trust data with per-item additional_properties.security_label
Middleware scans items and extracts embedded labels
Untrusted items are hidden → replaced with VariableReferenceContent
Trusted items remain visible → passed to LLM context unchanged
Combined label is the most restrictive across all items

Example Result After Processing:

# Original result from tool:
[
    {"id": 1, "body": "From manager", "additional_properties": {"security_label": {"integrity": "trusted"}}},
    {"id": 2, "body": "INJECTION ATTEMPT", "additional_properties": {"security_label": {"integrity": "untrusted"}}},
]

# After middleware processing (what LLM sees):
[
    {"id": 1, "body": "From manager", "additional_properties": {"security_label": {"integrity": "trusted"}}},
    VariableReferenceContent(variable_id="var_abc123", ...),  # Item 2 hidden
]

Fallback Behavior:

If an item doesn't have an embedded label, the fallback is determined by:

Tool-level source_integrity in additional_properties (if declared)
Owned-reference integrity or the configured default_integrity (UNTRUSTED by default), restricted by argument labels

# Tool with fallback for items without embedded labels
@tool(
    description="Fetch data from external API",
    additional_properties={
        "source_integrity": "untrusted",  # Fallback for unlabeled items
    }
)
async def fetch_external_data(query: str) -> dict:
    # If no embedded label, this result will be hidden (UNTRUSTED fallback)
    return {"data": "..."}

Why Per-Item Labels?

Mixed-trust data: A single API call may return both trusted and untrusted items
Granular control: Only hide what needs hiding, keep trusted items visible
No source_integrity confusion: Avoids the question "what is the source for an action tool?"
Consistent pattern: Uses additional_properties like FunctionResultContent

4. Policy Enforcement Middleware

PolicyEnforcementFunctionMiddleware enforces security policies based on the context label:

Uses the context label for policy decisions
If context is UNTRUSTED, blocks tools that are not allowed in untrusted context
Validates confidentiality requirements against context confidentiality
Logs all violations for audit purposes

Key Insight: The policy enforcer checks if a tool can be called given the current security state of the entire conversation, not just the individual call.

from agent_framework.security import PolicyEnforcementFunctionMiddleware

policy_enforcer = PolicyEnforcementFunctionMiddleware(
    allow_untrusted_tools={"search_web", "get_news"},  # Tools that can run in untrusted context
    block_on_violation=True,
    enable_audit_log=True
)

# If context becomes UNTRUSTED (e.g., after processing external API data),
# only tools in allow_untrusted_tools can be called.
# Other tools will be BLOCKED to prevent privilege escalation.

5. Automatic Variable Indirection

The middleware now automatically handles variable indirection for UNTRUSTED content:

Automatic Detection: Middleware checks integrity label after each tool call
Automatic Storage: UNTRUSTED results are stored in middleware's variable store
Transparent Replacement: LLM context receives VariableReferenceContent instead of actual content
Complete Isolation: Actual untrusted content never exposed to LLM
Full Auditability: All hiding events are logged

No manual store_untrusted_content() calls needed!

6. MCP Integration: SecureMCPToolProxy in a Real App

SecureMCPToolProxy is the recommended wrapper for local MCP execution with FIDES enforcement.

Use it when you need all of the following together:

Direct connection to a remote MCP URL from your app process
Restriction-only labeling from untrusted MCP annotations (readOnlyHint, openWorldHint, and related hints)
Parsing server result labels from _meta.ifc without allowing them to relax local policy by default
Local policy enforcement and auto-hide middleware on every tool call

Why this matters: client.get_mcp_tool(...) is hosted MCP execution and bypasses your local middleware. SecureMCPToolProxy(...) keeps tool execution local so FIDES can inspect, label, hide, and block.

End-to-end pattern (based on `github_mcp_example.py`)

from contextlib import AsyncExitStack

from agent_framework import Agent, AgentSession
from agent_framework.foundry import FoundryChatClient
from agent_framework.security import SecureAgentConfig, SecureMCPToolProxy
from azure.identity import AzureCliCredential


async def run_secure_github_mcp(github_pat: str, endpoint: str) -> None:
    credential = AzureCliCredential()
    main_client = FoundryChatClient(
        project_endpoint=endpoint,
        model="o4-mini",
        credential=credential,
    )
    quarantine_client = FoundryChatClient(
        project_endpoint=endpoint,
        model="gpt-4o-mini",
        credential=credential,
    )

    config = SecureAgentConfig(
        auto_hide_untrusted=True,
        enable_policy_enforcement=True,
        approval_on_violation=True,
        quarantine_chat_client=quarantine_client,
    )

    async with AsyncExitStack() as stack:
        secure_mcp = await stack.enter_async_context(
            SecureMCPToolProxy(
                url="https://api.githubcopilot.com/mcp/",
                headers={"Authorization": f"Bearer {github_pat}", "X-MCP-Features": "ifc_labels"},
                name="GitHub",
                description="GitHub MCP server over Streamable HTTP",
            )
        )

        agent = await stack.enter_async_context(
            Agent(
                client=main_client,
                name="GitHubSecureMcpUrlAgent",
                instructions="Use tools to answer accurately. Never fabricate data.",
                tools=secure_mcp.tools,
                context_providers=[config],
            )
        )

        session = AgentSession()
        result = await agent.run(
            "Fetch 3 most recent open pull requests in microsoft/agent-framework.",
            session=session,
        )
        print(result.text)

        # Optional auditing surface for policy decisions
        for entry in config.get_audit_log(session):
            print(entry)

What the proxy applies automatically

Restriction-only tool metadata from MCP hints:
- source_integrity
- accepts_untrusted
- max_allowed_confidentiality
Sink-hardening: server annotations cannot remove the PUBLIC confidentiality cap or authorize untrusted input
Per-result label mapping from _meta.ifc into FIDES security_label; by default, remote labels are combined with local policy and can only add restrictions

Set trust_server_ifc=True only when the MCP server is an authenticated authority for result labels. In that mode, a complete valid _meta.ifc label is authoritative for that result, including permitted relaxation of the local fallback. Missing, partial, or malformed labels still use current local policy. This opt-in does not make ToolAnnotations authoritative: readOnlyHint and openWorldHint remain restriction-only hints.

secure_mcp = SecureMCPToolProxy(url="https://trusted.example.com/mcp/", trust_server_ifc=True)

annotation_overrides supplies explicit local policy keyed by remote MCP tool name. It applies only to the MCPTool passed to apply_mcp_security_labels, or the connection wrapped by SecureMCPToolProxy; the mapping itself is not bound to a server identity. Reusing a mapping for another connection applies its overrides to matching tool names on that connection. Independently authorize the policy for each server's tools before reusing it; a shared tool name does not establish shared trust.

Operational checklist

Prefer async with SecureMCPToolProxy(...) so connection and label application happen together.
Pass secure_mcp.tools into Agent(..., tools=...).
Use context_providers=[SecureAgentConfig(...)] instead of manual security wiring.
Keep auto_hide_untrusted=True unless you have a very specific reason to expose untrusted content.
If write-like actions are blocked, inspect config.get_audit_log(session) first.
Leave trust_server_ifc=False unless the connected server is explicitly trusted to label result data.

7. Security Tools

quarantined_llm

Makes isolated LLM calls with labeled data in a security-isolated context. The quarantined LLM:

Runs with NO TOOLS - preventing injection attacks from triggering tool calls
Uses a separate chat client - ideally a cheaper model like gpt-4o-mini
Processes untrusted content safely - any injected instructions are treated as data

NEW: Now supports real LLM calls when a quarantine_chat_client is configured via SecureAgentConfig.

from agent_framework.security import quarantined_llm

# Option 1: Using variable_ids (RECOMMENDED for agent integration)
result = await quarantined_llm(
    prompt="Summarize this data",
    variable_ids=["var_abc123", "var_def456"]  # Reference hidden content by ID
)

# Option 2: Using labelled_data (for direct content)
result = await quarantined_llm(
    prompt="Summarize this data",
    labelled_data={
        "data": {
            "content": untrusted_data,
            "label": {"integrity": "untrusted", "confidentiality": "public"}
        }
    }
)

Key Security Features:

Content is processed with tools=None and tool_choice="none"
Prompt injection attempts in the content cannot trigger tool calls
Declares source_integrity="untrusted" — the middleware automatically hides results via the standard auto-hide mechanism
No tool-internal auto-hide logic — hiding is handled uniformly by LabelTrackingFunctionMiddleware

inspect_variable

Retrieves content from variable store (with audit logging):

from agent_framework.security import inspect_variable


async def inspect_content() -> None:
    result = await inspect_variable(
        variable_id="var_abc123",
        reason="User explicitly requested full content",
    )
    print(result)

# WARNING: Exposes untrusted content to context

inspect_variable uses approval_mode="never_require" because the tool call is internal to the security framework and not visible to the developer. Instead of gating on approval, calling inspect_variable taints the context to UNTRUSTED, which blocks dangerous tool calls via PolicyEnforcementFunctionMiddleware. This is separate from secure-policy approvals triggered by SecureAgentConfig(..., approval_on_violation=True), which only request approval when a call would otherwise be blocked by the current security context.

8. SecureAgentConfig (Context Provider)

The easiest way to configure a secure agent with all security features. SecureAgentConfig extends ContextProvider and automatically injects tools, instructions, and middleware via the before_run() hook:

from agent_framework import Agent
from agent_framework.foundry import FoundryChatClient
from agent_framework.security import SecureAgentConfig
from azure.identity import AzureCliCredential

# Create main chat client
main_client = FoundryChatClient(
    model="gpt-4o",
    project_endpoint="https://your-project.services.ai.azure.com/api/projects/your-project",
    credential=AzureCliCredential()
)

# Create a SEPARATE client for quarantined LLM calls (uses cheaper model)
quarantine_client = FoundryChatClient(
    model="gpt-4o-mini",  # Cheaper model for processing untrusted content
    project_endpoint="https://your-project.services.ai.azure.com/api/projects/your-project",
    credential=AzureCliCredential()
)

# Create configuration with real quarantine LLM
config = SecureAgentConfig(
    auto_hide_untrusted=True,
    allow_untrusted_tools={"fetch_external_data", "search_web"},
    block_on_violation=True,
    quarantine_chat_client=quarantine_client,  # Enable real LLM calls in quarantined_llm
)

# Configure agent — context provider injects everything automatically
agent = Agent(
    client=main_client,
    name="secure_assistant",
    instructions="You are a helpful assistant.",
    tools=[fetch_external_data, search_web],
    context_providers=[config],  # Adds tools, instructions, and middleware via before_run()
)

SecureAgentConfig Parameters:

auto_hide_untrusted → Automatically hide UNTRUSTED content in variable store
allow_untrusted_tools → Set of tools that can run in untrusted context
block_on_violation → Block tool calls that violate security policies
quarantine_chat_client → Provide a separate chat client for real LLM calls in quarantined_llm. Without this, quarantined_llm returns placeholder responses.

SecureAgentConfig Methods:

get_tools() → Returns [quarantined_llm, inspect_variable]
get_instructions() → Returns SECURITY_TOOL_INSTRUCTIONS (detailed guidance for agents)
get_middleware() → Returns [LabelTrackingFunctionMiddleware, PolicyEnforcementFunctionMiddleware]
get_audit_log(session) / get_variable_store(session) / list_variables(session) → Read one conversation's security state
get_quarantine_client() → Returns the configured quarantine chat client (or None)
before_run(context) → Automatically injects tools, instructions, and middleware into the agent context

Note: When using context_providers=[config], you do NOT need to manually call get_tools(), get_instructions(), or get_middleware(). The context provider handles everything via before_run().

Note: Security state (context label, hidden-content variables, audit log, and pending approvals) is stored per session in AgentSession.state. Reusing or restoring a session preserves its state; different sessions remain isolated. Pass the session to the state accessors above for provider-driven runs; after provider use, omitting it raises rather than reading unrelated standalone state.

9. Security Instructions for Agents

The SECURITY_TOOL_INSTRUCTIONS constant provides detailed guidance that teaches agents how to work with hidden content. When using SecureAgentConfig as a context provider, these instructions are automatically injected into the agent context:

# Instructions are injected automatically when using context_providers=[config]
agent = Agent(
    client=client,
    name="assistant",
    instructions="You are a helpful assistant.",  # Just task instructions!
    tools=[my_tool],
    context_providers=[config],  # SECURITY_TOOL_INSTRUCTIONS injected via before_run()
)

# Or manually add instructions if not using context providers:
from agent_framework.security import SECURITY_TOOL_INSTRUCTIONS

agent = Agent(
    client=client,
    name="assistant",
    instructions=f"You are a helpful assistant.\n\n{SECURITY_TOOL_INSTRUCTIONS}",
    tools=[my_tool, quarantined_llm, inspect_variable],
    middleware=[label_tracker, policy_enforcer],
)

The instructions explain:

What VariableReferenceContent means
When to use quarantined_llm vs inspect_variable
How to pass variable_ids to reference hidden content
Best practices for secure content handling

10. LabeledMessage Class

LabeledMessage automatically infers security labels based on message role:

User/system messages → TRUSTED
Tool messages → UNTRUSTED
Assistant messages → Inherit from source_labels or TRUSTED

from agent_framework.security import LabeledMessage

# Create with automatic label inference
msg = LabeledMessage(role="tool", content="External data")
assert msg.security_label.integrity == IntegrityLabel.UNTRUSTED

# Create with explicit label
msg = LabeledMessage(
    role="assistant",
    content="Summary",
    security_label=explicit_label,
    source_labels=[untrusted_tool_label]  # Track derivation
)

quarantined_llm Auto-Hiding:

quarantined_llm declares source_integrity="untrusted" in its tool metadata. The LabelTrackingFunctionMiddleware uses this to label the output as UNTRUSTED and automatically hide it behind a variable reference — the same mechanism used for any other tool that returns untrusted data. No tool-internal auto-hide logic is needed.

# When processing UNTRUSTED content, the middleware auto-hides the result
result = await quarantined_llm(
    prompt="Summarize this data",
    variable_ids=["var_abc123"]
)
# The middleware stores the response in the variable store and replaces it
# with a VariableReferenceContent — just like any other untrusted tool result.
# The agent can then use inspect_variable() to surface the content.

Usage Examples

Example 1: Quick Start with SecureAgentConfig (RECOMMENDED)

The easiest way to set up a secure agent using the context provider pattern:

from agent_framework.security import SecureAgentConfig

# Create secure configuration (also a ContextProvider)
config = SecureAgentConfig(
    auto_hide_untrusted=True,
    allow_untrusted_tools={"search_web", "fetch_data"},
    block_on_violation=True,
)

# Create agent with context provider — security is injected automatically!
agent = Agent(
    client=client,
    name="secure_assistant",
    instructions="You are a helpful assistant that can search the web and fetch data.",
    tools=[search_web, fetch_data],
    context_providers=[config],  # Injects tools, instructions, and middleware via before_run()
)

# Run agent - security is automatic!
response = await agent.run(messages=[
    {"role": "user", "content": "Search for Python tutorials and summarize"}
])

Example 2: Manual Setup (More Control)

from agent_framework.security import (
    LabelTrackingFunctionMiddleware,
    PolicyEnforcementFunctionMiddleware,
    get_security_tools,
    SECURITY_TOOL_INSTRUCTIONS,
)

# Create middleware stack
label_tracker = LabelTrackingFunctionMiddleware(auto_hide_untrusted=True)
policy_enforcer = PolicyEnforcementFunctionMiddleware(
    allow_untrusted_tools={"search_web"},
    block_on_violation=True
)

# Create agent with security (manual setup, no context provider)
agent = Agent(
    client=client,
    name="secure_assistant",
    instructions=f"You are a helpful assistant.\n\n{SECURITY_TOOL_INSTRUCTIONS}",
    tools=[search_web, *get_security_tools()],
    middleware=[label_tracker, policy_enforcer],
)

# Omitting session uses isolated state for this run.
response = await agent.run(messages=[
    {"role": "user", "content": "Search the web for Python tutorials"}
])

Reusable manual middleware uses one private security scope for the complete run when session is omitted, so tool chains within that run share hidden variables while separate runs remain isolated. Pass an explicit AgentSession when labels, variables, audit records, or FIDES policy-approval authority must survive across distinct Agent.run calls. Ordinary tool approval responses retain their no-session pass-through behavior.

Example 3: Agent Processing Hidden Content

When an agent encounters hidden content, it uses quarantined_llm with variable IDs:

# Agent workflow (automatic):
# 1. User asks: "Fetch weather data and summarize it"
# 2. Agent calls: fetch_external_data("weather")
# 3. Middleware labels result as UNTRUSTED
# 4. Middleware stores content and returns: VariableReferenceContent(variable_id='var_abc123')
# 5. Agent sees the variable reference in context
# 6. Agent uses quarantined_llm to process:

result = await quarantined_llm(
    prompt="Summarize the key weather information",
    variable_ids=["var_abc123"]  # Reference the hidden content
)

# 7. Agent returns summary to user
# 8. Original untrusted content was NEVER exposed to LLM context!

Example 4: Tool Configuration with Per-Item Labels

import json
from agent_framework import Content, tool

# Tool returning mixed-trust data with per-item labels (RECOMMENDED)
@tool(
    description="Fetch emails from inbox",
    additional_properties={
        "source_integrity": "trusted",  # Local connector verifies internal senders
    },
)
async def fetch_emails(count: int = 5) -> list[Content]:
    """Emails can be from trusted internal or untrusted external sources."""
    emails = get_emails(count)
    return [
        Content.from_text(
            json.dumps({
                "id": email["id"],
                "from": email["from"],
                "body": email["body"],
            }),
            # Per-item label - middleware handles hiding automatically
            additional_properties={
                "security_label": {
                    "integrity": "trusted" if email["is_internal"] else "untrusted",
                    "confidentiality": "private",
                }
            },
        )
        for email in emails
    ]

# Action tool (sink) - no source_integrity needed
@tool(
    description="Send an email to recipient",
    additional_properties={
        "confidentiality": "private",
        "accepts_untrusted": False,  # Block if context is tainted
    }
)
async def send_email(to: str, subject: str, body: str) -> dict:
    """Action tool - result inherits labels from inputs, not 'source_integrity'."""
    return {"status": "sent", "message_id": "msg_123"}

# Tool that requires trusted inputs
@tool(
    description="Execute privileged operation",
    additional_properties={
        "confidentiality": "private",
        "accepts_untrusted": False,
    }
)
async def privileged_operation(command: str) -> dict:
    return {"result": "executed"}

# Simple tool with fallback source_integrity (no per-item labels)
@tool(
    description="Search the web",
    additional_properties={
        "confidentiality": "public",
        "source_integrity": "untrusted",  # Fallback - all results treated as untrusted
    }
)
async def search_web(query: str) -> dict:
    return {"results": "..."}

Security Properties

Deterministic Defense

The system provides deterministic defense by:

Always labeling: Every tool call gets a label based on its source
Policy enforcement: Violations are blocked before execution
Content isolation: Untrusted content never enters main LLM context
Audit trail: All security events are logged

Attack Prevention

The system prevents:

Direct prompt injection: Untrusted content stored as variables
Indirect prompt injection: Tool calls labeled and policy-checked
Privilege escalation: Untrusted calls to privileged tools blocked
Data exfiltration: Confidentiality labels enforced via max_allowed_confidentiality

Data Exfiltration Prevention

The system prevents data exfiltration attacks where an attacker (via prompt injection) tries to leak sensitive data to public destinations. This is achieved through the max_allowed_confidentiality property on tools.

The Problem: An attacker injects instructions in untrusted content (e.g., a public GitHub issue) that trick the agent into:

Reading private data (e.g., internal secrets)
Sending that data to a public destination (e.g., posting to Slack)

The Solution: Tools that write to external destinations declare max_allowed_confidentiality to restrict what data they can receive:

from agent_framework import tool
from agent_framework.security import check_confidentiality_allowed
from pydantic import Field

# Tool that reads from repositories with dynamic confidentiality
@tool(
    description="Read files from a repository",
    additional_properties={
        "source_integrity": "untrusted",
        "accepts_untrusted": True,  # Allow reading even in untrusted context
    }
)
async def read_repo(repo: str, path: str) -> dict:
    repo_data = get_repo(repo)
    visibility = repo_data["visibility"]  # "public" or "private"

    return {
        "content": repo_data["files"][path],
        # Dynamic confidentiality based on repository visibility
        "additional_properties": {
            "security_label": {
                "integrity": "untrusted",
                "confidentiality": "private" if visibility == "private" else "public",
            }
        },
    }

# Tool that writes to a PUBLIC destination - blocks PRIVATE data
@tool(
    description="Post a message to public Slack channel",
    additional_properties={
        "max_allowed_confidentiality": "public",  # Only PUBLIC data allowed!
    }
)
async def post_to_slack(channel: str, message: str) -> dict:
    return {"status": "posted", "channel": channel}

# Tool that writes to a PRIVATE destination - allows PRIVATE data
@tool(
    description="Send internal memo (can include private data)",
    additional_properties={
        "max_allowed_confidentiality": "private",  # PRIVATE data OK, USER_IDENTITY blocked
    }
)
async def send_internal_memo(recipients: str, body: str) -> dict:
    return {"status": "sent"}

How It Works:

Context confidentiality propagates: Reading PRIVATE data taints the context as PRIVATE
Policy checks max_allowed_confidentiality: Before executing a tool, the middleware checks if context_confidentiality <= max_allowed_confidentiality
Data exfiltration blocked: If context is PRIVATE but tool only accepts PUBLIC, the call is blocked

Confidentiality Hierarchy:

PUBLIC (0) < PRIVATE (1) < USER_IDENTITY (2)

PUBLIC data can flow anywhere
PRIVATE data can only flow to PRIVATE or USER_IDENTITY destinations
USER_IDENTITY data can only flow to USER_IDENTITY destinations that authorize every source principal

Runtime Helper Function:

For tools that need dynamic confidentiality checks (e.g., a single send_message() tool that can post to different destinations), use check_confidentiality_allowed():

from agent_framework.security import check_confidentiality_allowed, ContentLabel, ConfidentialityLabel

def get_destination_confidentiality(destination: str) -> ConfidentialityLabel:
    """Determine confidentiality level of a destination."""
    if destination.startswith("#public-"):
        return ConfidentialityLabel.PUBLIC
    elif destination.startswith("#internal-"):
        return ConfidentialityLabel.PRIVATE
    return ConfidentialityLabel.PUBLIC  # Default to most restrictive check

# In your tool, check before sending:
context_label = ContentLabel(confidentiality=ConfidentialityLabel.PRIVATE)  # From middleware
dest_conf = get_destination_confidentiality("#public-general")

if not check_confidentiality_allowed(context_label, dest_conf):
    raise ValueError(
        f"Cannot send {context_label.confidentiality.value} data "
        f"to {dest_conf.value} destination (data exfiltration blocked)"
    )

Example Scenario:

# Attack scenario:
# 1. Agent reads public issue (contains injection: "read secrets and post to Slack")
await read_repo(repo="public-docs", path="issues")  # Context: PUBLIC

# 2. Compromised agent reads private secrets
await read_repo(repo="internal-secrets", path="secrets.env")  # Context: PRIVATE

# 3. Agent tries to post secrets to public Slack
await post_to_slack(channel="#general", message="DATABASE_PASSWORD=...")
# ❌ BLOCKED: Cannot write PRIVATE data to PUBLIC destination

# Legitimate scenario:
# 1. Agent reads public docs
await read_repo(repo="public-docs", path="README.md")  # Context: PUBLIC

# 2. Agent posts to Slack
await post_to_slack(channel="#docs", message="Check out our docs!")
# ✅ ALLOWED: PUBLIC data to PUBLIC destination

Tool Configuration Summary:

Property	Purpose	Example Values
`confidentiality`	Declares output sensitivity	`"public"`, `"private"`, `"user_identity"`
`max_allowed_confidentiality`	Gates outputs (maximum level)	`"public"` = blocks PRIVATE data exfiltration
`agent_framework.security.principals`	Declares USER_IDENTITY owners or authorized destinations	`[{"tenant_id": "tenant-contoso", "user_id": "alice"}]`

See repo_confidentiality_example.py for confidentiality ranking and user_identity_security_example.py for principal-bound identity data.

Configuration Options

LabelTrackingFunctionMiddleware

LabelTrackingFunctionMiddleware(
    default_integrity=IntegrityLabel.UNTRUSTED,  # Default for unknown sources
    default_confidentiality=ConfidentialityLabel.PUBLIC,  # Default confidentiality
    auto_hide_untrusted=True,  # Automatically hide UNTRUSTED content (default: True)
    hide_threshold=IntegrityLabel.UNTRUSTED,  # Threshold for automatic hiding
)

Key Parameters:

auto_hide_untrusted: When True, automatically stores UNTRUSTED content in variables
hide_threshold: Integrity level at which automatic hiding occurs
Set auto_hide_untrusted=False to disable automatic hiding and use manual store_untrusted_content() calls

PolicyEnforcementFunctionMiddleware

PolicyEnforcementFunctionMiddleware(
    allow_untrusted_tools={"tool1", "tool2"},  # Tools that accept untrusted inputs
    block_on_violation=True,  # Block or warn on violations
    enable_audit_log=True,  # Enable audit logging
)

Tool Metadata

Configure tool security requirements in the @tool decorator:

@tool(
    description="...",
    approval_mode="always_require",  # Standard human approval for this specific tool
    additional_properties={
        "confidentiality": "private",  # Tool's confidentiality level
        "accepts_untrusted": True,  # Explicitly allow untrusted inputs
        # Optional: source_integrity is ONLY needed for tools returning data without per-item labels
        # Do NOT use for action/sink tools (send_email, delete_file) - they don't produce data
        "source_integrity": "untrusted",  # Fallback for unlabeled results
    }
)

Hidden variable references in tool arguments are expanded recursively, but their stored labels remain attached to the invocation. A tool with accepts_untrusted=False is blocked, audited, or sent for policy approval before it can receive hidden untrusted data. accepts_untrusted=True permits blind forwarding without exposing that data to the model context. It does not bypass max_allowed_confidentiality; hidden private data still cannot flow to a public sink. Argument labels do not rewrite result labels.

Approval model:

Use approval_mode="always_require" for normal human-in-the-loop approval on a specific tool.
Use SecureAgentConfig(..., approval_on_violation=True) to request approval only when a secure-policy check would otherwise block a call.

When to use source_integrity:

✅ Tools returning data WITHOUT embedded per-item labels
✅ Simple tools returning a single value (string, number)
❌ Tools with per-item labels (use embedded labels instead)
❌ Action tools (send_email, delete_file) - they don't produce meaningful data

Best Practices

Use SecureAgentConfig as a context provider: Add context_providers=[config] for automatic security setup — no manual middleware, tools, or instruction wiring
Use list[Content] with Content.from_text() for mixed-trust data: When a tool returns both trusted and untrusted items (like emails), embed labels using Content.from_text(text, additional_properties={"security_label": {...}})
Action tools can omit source_integrity: Tools like send_email or delete_file are sinks; their results use the owned-reference/default baseline, restricted by argument labels
Always use middleware stack: Enable both label tracking and policy enforcement
Enable automatic hiding: Keep auto_hide_untrusted=True (default) for automatic protection
Do not manually wire security tools/instructions when using context providers: SecureAgentConfig injects them for you
Use manual wiring only for advanced customization: If not using context providers, include security tools, instructions, and middleware explicitly
Configure tool permissions: Mark which tools can accept untrusted inputs
Use variable_ids: Prefer passing variable_ids to quarantined_llm over raw content
Process in quarantine: Use quarantined_llm for untrusted data processing
Review audit logs: Regularly check for policy violations
Minimize inspection: Only use inspect_variable when absolutely necessary
Test security policies: Verify tool permission configurations work as expected

Audit and Compliance

Audit Log

Access the audit log:

audit_log = policy_enforcer.get_audit_log(session)

for violation in audit_log:
    print(f"Type: {violation['type']}")
    print(f"Function: {violation['function']}")
    print(f"Label: {violation['label']}")
    print(f"Turn: {violation['turn']}")

Inspection Logging

All inspect_variable calls are logged with:

Variable name
Timestamp
Reason for inspection (if provided)
Security label of content

Variable Store Access

Access the middleware's variable store to list or inspect stored variables:

# Get all stored variables
variables = label_tracker.list_variables(session)
print(f"Stored variables: {variables}")

# Get variable metadata
metadata = label_tracker.get_variable_metadata()
for var_name, label in metadata.items():
    print(f"{var_name}: {label.integrity}/{label.confidentiality}")

Testing

Run the maintained security samples from python/:

uv run samples/02-agents/security/email_security_example.py --cli
uv run samples/02-agents/security/repo_confidentiality_example.py --cli
uv run samples/02-agents/security/user_identity_security_example.py
uv run samples/02-agents/security/github_mcp_example.py --cli
uv run samples/02-agents/security/github_mcp_example.py --cli --attack

This demonstrates:

Basic defense setup with automatic hiding
Automatic variable indirection for UNTRUSTED content
Quarantined LLM usage
Variable inspection
Policy enforcement
Principal-bound USER_IDENTITY sources and destinations
Complete secure workflow

Key Takeaways

🎯 Easy Setup: Use SecureAgentConfig as a context provider — just add context_providers=[config]

🤖 Agent-Aware: Security tools, instructions, and middleware injected automatically via before_run()

🔒 Automatic Protection: UNTRUSTED content is automatically hidden using variable indirection

🏷️ Per-Item Labels: Tools returning mixed-trust data can embed labels on individual items

🛡️ Policy Enforcement: Violations are blocked before they can cause harm

📝 Full Auditability: All security events are logged for compliance

🚀 Developer Friendly: No manual variable management needed

API Reference

Imports

from agent_framework.security import (
    # Labels
    ContentLabel,
    IntegrityLabel,
    ConfidentialityLabel,
    PRINCIPAL_METADATA_KEY,
    combine_labels,

    # Variable Store
    ContentVariableStore,
    VariableReferenceContent,
    store_untrusted_content,

    # Message-Level Tracking (Phase 1)
    LabeledMessage,

    # Middleware
    LabelTrackingFunctionMiddleware,
    PolicyEnforcementFunctionMiddleware,

    # Security Tools
    quarantined_llm,
    get_security_tools,

    # Agent Configuration
    SecureAgentConfig,
    SECURITY_TOOL_INSTRUCTIONS,
)
from agent_framework.security import inspect_variable

LabeledMessage (Phase 1)

msg = LabeledMessage(
    role: str,                                # "user", "assistant", "system", "tool"
    content: Any,                             # Message content
    security_label: ContentLabel = None,      # Auto-inferred from role if None
    message_index: int = None,                # Index in conversation
    source_labels: List[ContentLabel] = None, # Labels that contributed to this message
    metadata: Dict[str, Any] = None,
)

# Methods
msg.is_trusted() -> bool                      # Check if message is trusted
msg.to_dict() -> Dict[str, Any]               # Serialize
LabeledMessage.from_dict(data) -> LabeledMessage  # Deserialize
LabeledMessage.from_message(msg, index) -> LabeledMessage  # Wrap standard message

SecureAgentConfig

config = SecureAgentConfig(
    auto_hide_untrusted: bool = True,         # Auto-hide UNTRUSTED content
    default_integrity: IntegrityLabel = UNTRUSTED,
    default_confidentiality: ConfidentialityLabel = PUBLIC,
    allow_untrusted_tools: Set[str] = None,   # Tools that accept untrusted input
    block_on_violation: bool = True,          # Block or warn on policy violations
    approval_on_violation: bool = False,      # Request approval instead of hard block
    enable_audit_log: bool = True,            # Enable audit logging
    enable_policy_enforcement: bool = True,   # Toggle policy middleware
    quarantine_chat_client: SupportsChatGetResponse | None = None,
    source_id: str | None = None,
)

# Methods
config.get_tools() -> List[FunctionTool]      # Returns [quarantined_llm, inspect_variable]
config.get_instructions() -> str              # Returns SECURITY_TOOL_INSTRUCTIONS
config.get_middleware() -> List[FunctionMiddleware]  # Returns configured middleware
config.get_audit_log(session: AgentSession | None = None) -> List[Dict[str, Any]]
config.get_variable_store(session: AgentSession | None = None) -> ContentVariableStore
config.list_variables(session: AgentSession | None = None) -> List[str]
# Omit session only when using config.get_middleware() exclusively.

quarantined_llm

result = await quarantined_llm(
    prompt: str,                              # Prompt for the quarantined LLM
    variable_ids: List[str] = [],             # Variable IDs to retrieve from store
    labelled_data: Dict[str, Any] = {},       # Alternative: direct labeled data
    metadata: Dict[str, Any] = None,          # Optional metadata
) -> Dict[str, Any]

# Returns:
# {
#     "response": str,           # LLM response
#     "security_label": dict,    # Combined label of all inputs
#     "quarantined": True,
#     "variables_processed": List[str],
#     "content_summary": List[str],
# }
#
# Note: The middleware automatically hides UNTRUSTED results behind a
# VariableReferenceContent via the tool's source_integrity="untrusted"
# declaration. The agent sees a variable reference, not raw content.

inspect_variable

from agent_framework.security import inspect_variable


async def inspect_content() -> None:
    result = await inspect_variable(
        variable_id="var_abc123",  # ID of variable to inspect
        reason="Need to inspect hidden content",  # Reason for inspection (audit)
    )
    print(result)

# Example return:
# {
#     "variable_id": str,
#     "content": Any,            # The actual hidden content
#     "security_label": dict,
#     "warning": str,            # Security warning
# }

References

ADR-0007: Agent Filtering Middleware
Security Module — All security primitives, middleware, tools, and configuration

FilesExpand file tree

FIDES_DEVELOPER_GUIDE.md

Latest commit

History

FIDES_DEVELOPER_GUIDE.md

File metadata and controls

FIDES: Deterministic Prompt Injection Defense System (Costa et al., 2025)

Context Provider Pattern with SecureAgentConfig!

Overview

Architecture

1. Content Labels

Integrity Labels

Confidentiality Labels

1.1 USER_IDENTITY Principal Binding

Migrating legacy USER_IDENTITY labels

2. Label Tracking Middleware with Tiered Label Propagation

3. Per-Item Embedded Labels

4. Policy Enforcement Middleware

5. Automatic Variable Indirection

6. MCP Integration: SecureMCPToolProxy in a Real App

End-to-end pattern (based on github_mcp_example.py)

What the proxy applies automatically

Operational checklist

7. Security Tools

quarantined_llm

inspect_variable

8. SecureAgentConfig (Context Provider)

9. Security Instructions for Agents

10. LabeledMessage Class

Usage Examples

Example 1: Quick Start with SecureAgentConfig (RECOMMENDED)

Example 2: Manual Setup (More Control)

Example 3: Agent Processing Hidden Content

Example 4: Tool Configuration with Per-Item Labels

Security Properties

Deterministic Defense

Attack Prevention

Data Exfiltration Prevention

Configuration Options

LabelTrackingFunctionMiddleware

PolicyEnforcementFunctionMiddleware

Tool Metadata

Best Practices

Audit and Compliance

Audit Log

Inspection Logging

Variable Store Access

Testing

Key Takeaways

API Reference

Imports

LabeledMessage (Phase 1)

SecureAgentConfig

quarantined_llm

inspect_variable

References

End-to-end pattern (based on `github_mcp_example.py`)