Skip to content

Azure AI Foundry Agents Integration

Swap AgentsClient for MendGuardrailsAzureFoundryClient. Everything else stays the same.

import asyncio
from azure.identity import DefaultAzureCredential
from mendguardrails import MendGuardrailsAzureFoundryClient, GuardrailEnforcementTriggered

async def main():
    async with MendGuardrailsAzureFoundryClient(
        endpoint="https://your-project.services.ai.azure.com/",
        credential=DefaultAzureCredential(),
        config="guardrails_config.json",
        offline=True,
    ) as client:
        thread = await client.threads.create()
        await client.messages.create(
            thread_id=thread.id,
            role="user",
            content="Summarise the Q2 security report.",
        )

        try:
            run = await client.runs.create_and_process(
                thread_id=thread.id,
                agent_id="your-agent-id",
            )
            print(run.status)  # real ThreadRun — type unchanged

        except GuardrailEnforcementTriggered as exc:
            print(f"Blocked by: {exc.guardrail_result.info['guardrail_name']}")

asyncio.run(main())

Installation

pip install "mend-guardrails[azure-foundry-agents]"

The extra pulls in azure-ai-agents. The base package does not depend on it so existing installations are unaffected.

How it works

MendGuardrailsAzureFoundryClient inherits directly from azure.ai.agents.aio.AgentsClient. All isinstance checks pass and every SDK method (threads, messages, runs.create, runs.stream, …) is available unchanged. Only runs.create_and_process and create_thread_and_process_run are intercepted to run guardrails.

The guardrail pipeline runs in three stages per request:

Stage When it runs Typical use
pre_flight Before the run starts, concurrently with input Jailbreak, PII masking
input Before the run starts, concurrently with pre_flight Prompt injection detection, harmful-content filter
output After the run completes Prompt injection in tool output, secret-key leakage, URL filter

pre_flight and input run concurrently. output runs after the LLM responds. The first failing guardrail raises GuardrailEnforcementTriggered and stops the pipeline immediately.

flowchart TD
    A([User request]) --> B & C
    B["pre_flight stage\nJailbreak · PII masking"]
    C["input stage\nPrompt injection · Harmful content"]
    B & C --> D{Any guardrail\nfails?}
    D -- Yes --> E([GuardrailEnforcementTriggered])
    D -- No --> F["Azure AI Foundry run\ncreate_and_process"]
    F --> G["output stage\nPrompt injection · Secret keys · URL filter"]
    G --> H{Guardrail\nfails?}
    H -- Yes --> E
    H -- No --> I([Response returned to caller])

Convenience method — create_thread_and_process_run

If you create the thread and run in one call, guardrails are still applied:

from azure.ai.agents.models import AgentThreadCreationOptions, ThreadMessageOptions, MessageRole

thread_options = AgentThreadCreationOptions(
    messages=[
        ThreadMessageOptions(
            role=MessageRole.USER,
            content="What is in the Q2 report?",
        )
    ]
)

async with MendGuardrailsAzureFoundryClient(
    endpoint=endpoint,
    credential=DefaultAzureCredential(),
    config="guardrails_config.json",
    offline=True,
) as client:
    try:
        run = await client.create_thread_and_process_run(
            agent_id="your-agent-id",
            thread=thread_options,
        )
        print(run.status)
    except GuardrailEnforcementTriggered as exc:
        print(f"Blocked: {exc}")

Synchronous client

Use MendGuardrailsAzureFoundrySyncClient when you cannot use async/await:

from azure.identity import DefaultAzureCredential
from mendguardrails import MendGuardrailsAzureFoundrySyncClient, GuardrailEnforcementTriggered

with MendGuardrailsAzureFoundrySyncClient(
    endpoint="https://your-project.services.ai.azure.com/",
    credential=DefaultAzureCredential(),
    config="guardrails_config.json",
    offline=True,
) as client:
    thread = client.threads.create()
    client.messages.create(thread_id=thread.id, role="user", content="Hello")

    try:
        run = client.runs.create_and_process(thread_id=thread.id, agent_id="your-agent-id")
        print(run.status)
    except GuardrailEnforcementTriggered as exc:
        print(f"Blocked: {exc}")

Configuration options

Both clients accept the same configuration formats as the rest of the SDK:

from pathlib import Path
from mendguardrails import MendGuardrailsAzureFoundryClient, JsonString

# File path (recommended for production)
client = MendGuardrailsAzureFoundryClient(
    endpoint=endpoint,
    credential=credential,
    config=Path("guardrails_config.json"),
    offline=True,
)

# Dictionary (for dynamic / programmatic configuration)
config_dict = {
    "version": 1,
    "pre_flight": {"version": 1, "guardrails": [{"type": "jailbreak"}]},
    "input":  {"version": 1, "guardrails": [{"type": "prompt_injection_detection"}]},
    "output": {"version": 1, "guardrails": [{"type": "prompt_injection_detection"}]},
}
client = MendGuardrailsAzureFoundryClient(
    endpoint=endpoint,
    credential=credential,
    config=config_dict,
    offline=True,
)

# JSON string (with JsonString wrapper)
client = MendGuardrailsAzureFoundryClient(
    endpoint=endpoint,
    credential=credential,
    config=JsonString('{"version": 1, ...}'),
    offline=True,
)

Online mode

Omit config and offline to pull the policy from the Mend platform at runtime:

client = MendGuardrailsAzureFoundryClient(
    endpoint=endpoint,
    credential=credential,
    mend_key="your-mend-key",  # or set MEND_KEY env var
)

Audit events and callbacks

Pass on_guardrail_event to receive a callback for every guardrail result, whether it blocks or passes:

from mendguardrails.utils.events import GuardrailEvent

def handle_event(event: GuardrailEvent) -> None:
    print(f"[{event.stage}] {event.guardrail_name}: {event.result}")

client = MendGuardrailsAzureFoundryClient(
    endpoint=endpoint,
    credential=credential,
    config="guardrails_config.json",
    offline=True,
    on_guardrail_event=handle_event,
)

Multi-agent / A2A demo

The repository includes a full multi-agent demo that places guardrails at every A2A trust boundary:

sequenceDiagram
    actor User
    participant OA as OrchestratorAgent
    participant RA as ResearchAgent
    participant KB as Knowledge Base

    User->>OA: request (A2A)
    Note over OA: pre_flight — Jailbreak · PII masking

    OA->>RA: delegated task (A2A)
    Note over RA: input — PromptInjection · HarmfulContent

    RA->>KB: fetch_document tool
    KB-->>RA: document content
    Note over RA: output — PromptInjection · SecretKeys

    RA-->>OA: research result
    Note over OA: output — PromptInjection · URL Filter

    OA-->>User: final response

Requirements:

pip install "mend-guardrails[azure-foundry-agents]" "a2a-sdk" uvicorn httpx

Environment variables:

export AZURE_AI_FOUNDRY_PROJECT_ENDPOINT=https://your-project.services.ai.azure.com/
export AZURE_AI_AGENT_MODEL_DEPLOYMENT_NAME=gpt-4o
export MEND_KEY=your-mend-key

Run:

python -m examples.azure_foundry_a2a_demo

The demo exercises four security scenarios: a happy path, jailbreak detection, PII masking, and tool-output injection — all enforced automatically at the appropriate trust boundary.

Next steps