Azure AI Foundry

HighflameFoundryMiddleware for Azure AI Foundry — wraps the polling loop to guard user messages, tool calls, tool outputs, and assistant responses. Available for Python and TypeScript.

HighflameFoundryMiddleware integrates Highflame Shield into Azure AI Foundry agents. Unlike framework-level middleware APIs, Azure AI Foundry has no native hook registration — the caller owns the polling loop. This middleware wraps that loop, applying guardrail checks at four points:

• User prompt — last user message in the thread before the run starts

• Tool calls — each call at requires_action before execution

• Tool outputs — each result before submitting back to the model

• Assistant response — final assistant message after completed

On deny, raises BlockedError.

Available for Python and TypeScript.

Installation

Python

pip install 'highflame[foundry]'

TypeScript

Basic Usage (Polling)

The middleware owns the polling loop. Call createAndPollRun / acreate_and_poll_run in place of the raw Azure SDK polling.

Python

from highflame import Highflame
from highflame.integrations.foundry import HighflameFoundryMiddleware

client = Highflame(api_key="hf_sk_...")
middleware = HighflameFoundryMiddleware(client, mode="enforce")

# Add user message first
await project.agents.create_message(thread_id, {"role": "user", "content": user_input})

# Guard user message, then run with guardrails throughout
await middleware.aguard_user_message(project.agents, thread_id)
run = await middleware.acreate_and_poll_run(
    project.agents, thread_id, agent_id,
    execute_tools=my_tool_executor,
)

TypeScript

Basic Usage (Streaming)

Python

stream = project.agents.create_run_stream(thread_id, agent_id)
async for event in middleware.aprocess_stream(project.agents, thread_id, stream):
    handle(event)

TypeScript

All original stream events are re-yielded unchanged so the caller can still react to them. Guardrail checks run inline at thread.run.requires_action and thread.run.completed events.

Constructor

Python

HighflameFoundryMiddleware(
    client: Highflame,
    *,
    mode: str = "enforce",
    session_id: str | None = None,
    poll_interval: float = 1.0,
)

Parameter	Type	Default	Description
client	Highflame	required	Initialized Highflame client
mode	str	"enforce"	Enforcement mode: "enforce", "monitor", or "alert"
session_id	str | None	None	Static session ID. If not set, defaults to thread_id.
poll_interval	float	1.0	Seconds between polls in the run loop.

TypeScript

Session ID

When no session_id / sessionId is provided, the middleware uses the thread_id / threadId as the session ID. This is typically the right default — it groups all guardrail decisions for a conversation thread under one session.

Methods

guard_user_message / guardUserMessage

Fetches the last user message in the thread and evaluates it as a prompt. Call this after adding the user message and before creating the run.

Python

# Sync
middleware.guard_user_message(agents, thread_id)

# Async
await middleware.aguard_user_message(agents, thread_id)

TypeScript

create_and_poll_run / createAndPollRun

Creates a run and polls it to completion, applying guardrail checks at every interception point. Returns the final run object with status == "completed" (or the terminal status on failure).

Python

# Sync
run = middleware.create_and_poll_run(
    agents, thread_id, agent_id,
    execute_tools=my_tool_executor,   # optional; omit for runs with no tools
    poll_interval=2.0,                # optional; overrides constructor default
)

# Async
run = await middleware.acreate_and_poll_run(
    agents, thread_id, agent_id,
    execute_tools=my_async_tool_executor,
)

execute_tools receives the list of Azure tool call objects and must return the list of tool output objects. When omitted, empty string outputs are submitted so the run can complete.

TypeScript

process_stream / processStream

Processes a streaming run, guarding inline at event boundaries. Yields every original stream event unchanged.

Python

stream = agents.create_run_stream(thread_id, agent_id)
async for event in middleware.aprocess_stream(agents, thread_id, stream, execute_tools=my_executor):
    handle_event(event)

aprocess_stream is an async generator. There is no sync variant for streaming.

TypeScript

Guard Points

Point	When	Guard call
User prompt	Before run creation	evaluate_prompt / evaluatePrompt
Tool call (pre-execution)	At requires_action, before calling execute_tools	evaluate_tool_call / evaluateToolCall
Tool output (post-execution)	After execute_tools returns, before submit_tool_outputs	evaluate with content_type: "response"
Assistant response	After completed	evaluate with content_type: "response"

Enforcement Modes

Python

middleware = HighflameFoundryMiddleware(client, mode="enforce")  # block violations (default)
middleware = HighflameFoundryMiddleware(client, mode="monitor")  # observe without blocking
middleware = HighflameFoundryMiddleware(client, mode="alert")    # allow but trigger alerting

TypeScript

Complete Example

Python

import asyncio
from azure.ai.projects.aio import AIProjectClient
from azure.identity.aio import DefaultAzureCredential

from highflame import Highflame, BlockedError
from highflame.integrations.foundry import HighflameFoundryMiddleware

hf_client = Highflame(api_key="hf_sk_...")
middleware = HighflameFoundryMiddleware(hf_client, mode="enforce")

credential = DefaultAzureCredential()
project = AIProjectClient(endpoint=PROJECT_ENDPOINT, credential=credential)

async def run_agent(user_input: str, thread_id: str):
    try:
        await project.agents.create_message(
            thread_id, {"role": "user", "content": user_input}
        )
        await middleware.aguard_user_message(project.agents, thread_id)

        run = await middleware.acreate_and_poll_run(
            project.agents, thread_id, AGENT_ID,
            execute_tools=my_tool_executor,
        )

        if run.status == "completed":
            messages = await project.agents.list_messages(thread_id)
            return messages.data[0].content
        else:
            return {"error": f"Run ended with status: {run.status}"}

    except BlockedError as e:
        return {"error": f"Blocked: {e.response.policy_reason}"}

asyncio.run(run_agent("Summarize last quarter's sales data", "thread-abc-123"))

TypeScript

Error Handling

Python

from highflame import BlockedError

try:
    run = await middleware.acreate_and_poll_run(agents, thread_id, agent_id)
except BlockedError as e:
    print(f"Blocked at: {e.response.action}")
    print(f"Reason: {e.response.policy_reason}")
    print(f"Signals: {[s.name for s in e.response.signals]}")

TypeScript

Requirements

Python

Package	Minimum Version
azure-ai-projects	1.0.0+
azure-identity	1.15+
highflame	latest

TypeScript