DEV Community: Jubin Soni

A Deep Dive into the Microsoft Foundry Document Intelligence SDK: From PDF to Structured Data

Jubin Soni — Tue, 15 Sep 2026 18:21:01 +0000

Most "RAG over PDFs" pipelines have a step nobody talks about much: something has to turn a scanned invoice, a multi-column contract, or a photographed receipt into text a model can actually reason over. On Microsoft's stack, that something is usually the Document Intelligence SDK, formerly Form Recognizer, and it's worth understanding on its own terms rather than treating it as a black box that happens before the interesting part starts.

This is a hands-on deep dive into that SDK specifically. Not a tour of every Foundry Tools SDK, Vision and Speech and Content Safety each deserve their own treatment, but a real build using Document Intelligence: extracting layout as clean markdown, pulling structured fields out of a known document type, classifying documents before routing them, and training a custom extraction model on your own labeled data.

The mental model first

Two clients, and three kinds of model, cover almost everything this SDK does:

DocumentIntelligenceClient runs analysis. Every call goes through one method, begin_analyze_document, and a model_id parameter decides what kind of analysis happens. It's a long-running operation, so every call returns a poller.
DocumentIntelligenceAdministrationClient manages models. This is where you build custom extraction models and classifiers, list what's already been trained, and delete what you don't need anymore.
Prebuilt models (prebuilt-layout, prebuilt-invoice, prebuilt-receipt, prebuilt-idDocument, prebuilt-read, and others) handle common, well-known document shapes out of the box. No training required.
Custom extraction models, trained on your own labeled documents, handle document types nobody prebuilt a model for, your specific contract template, your specific intake form.
Classifiers solve a different problem entirely: given a document of unknown type, which model should even look at it? This matters more than it sounds like it should, since most real document pipelines receive a mix of types, not one known shape.

Prerequisites

A Document Intelligence resource (or a multi-service Foundry resource, which includes it), giving you an endpoint and either an API key or Entra ID access.
Python 3.9+ with the SDK installed.

pip install azure-ai-documentintelligence azure-identity

from azure.ai.documentintelligence import DocumentIntelligenceClient
from azure.core.credentials import AzureKeyCredential

endpoint = "https://YOUR-RESOURCE.cognitiveservices.azure.com"
client = DocumentIntelligenceClient(endpoint=endpoint, credential=AzureKeyCredential("YOUR-KEY"))

For anything past local experimentation, swap the key for DefaultAzureCredential and an RBAC role scoped to the resource, the same pattern every other Foundry-adjacent SDK in this series has used.

Step 1: layout extraction, straight to markdown

This is the single most useful call in the whole SDK if your end goal is feeding documents into a RAG pipeline. prebuilt-layout doesn't just extract text, it understands headings, tables, and section structure, and it can hand all of that back as GitHub-flavored markdown instead of a flat text blob.

from azure.ai.documentintelligence.models import AnalyzeDocumentRequest, DocumentContentFormat

with open("contract.pdf", "rb") as f:
    poller = client.begin_analyze_document(
        "prebuilt-layout",
        AnalyzeDocumentRequest(bytes_source=f.read()),
        output_content_format=DocumentContentFormat.MARKDOWN,
    )
result = poller.result()
print(result.content[:500])

result.content is now a markdown string, headings as #, tables as GFM pipe tables, page structure preserved. That matters more than it sounds like it should: a table flattened into plain text loses its row and column relationships, and a model reasoning over that text has to reconstruct structure it was never actually given. Markdown output keeps the structure intact.

Step 2: pulling structured fields from a known document type

For document types Document Intelligence already knows, invoices are the clearest example, you get named fields back with a confidence score per field, not just raw text.

with open("invoice.pdf", "rb") as f:
    poller = client.begin_analyze_document("prebuilt-invoice", AnalyzeDocumentRequest(bytes_source=f.read()))
result = poller.result()

for doc in result.documents:
    vendor = doc.fields.get("VendorName")
    total = doc.fields.get("InvoiceTotal")
    if vendor:
        print(f"Vendor: {vendor.value_string} (confidence: {vendor.confidence:.2f})")
    if total:
        print(f"Total: {total.value_currency.amount} (confidence: {total.confidence:.2f})")

That confidence score isn't decoration. It's the field you should actually branch on in production code, more on that in the production section below.

Step 3: add-on capabilities you'll want more often than the docs suggest

A few optional capabilities aren't on by default, since they add processing cost, but are worth turning on deliberately rather than discovering you needed them after the fact:

from azure.ai.documentintelligence.models import AnalyzeDocumentRequest, DocumentAnalysisFeature

with open("shipping-label.pdf", "rb") as f:
    poller = client.begin_analyze_document(
        "prebuilt-layout",
        AnalyzeDocumentRequest(bytes_source=f.read()),
        features=[DocumentAnalysisFeature.BARCODES, DocumentAnalysisFeature.FORMULAS],
    )

BARCODES extracts barcode and QR code payloads directly, useful for shipping labels and inventory documents where the barcode carries the actual identifier the text doesn't repeat. FORMULAS pulls out mathematical expressions as LaTeX, relevant if you're processing scientific or financial documents where a formula matters more than the surrounding prose. There's also a high-resolution mode for documents where small print matters, at the cost of slower processing.

Step 4: build a classifier to route mixed document types

Real intake pipelines rarely receive one document type. A classifier solves the "what am I even looking at" problem before you commit to an extraction model.

from azure.ai.documentintelligence import DocumentIntelligenceAdministrationClient
from azure.ai.documentintelligence.models import (
    BuildDocumentClassifierRequest,
    ClassifierDocumentTypeDetails,
    AzureBlobContentSource,
)

admin_client = DocumentIntelligenceAdministrationClient(endpoint=endpoint, credential=AzureKeyCredential("YOUR-KEY"))

poller = admin_client.begin_build_classifier(
    BuildDocumentClassifierRequest(
        classifier_id="support-doc-classifier",
        doc_types={
            "invoice": ClassifierDocumentTypeDetails(
                azure_blob_source=AzureBlobContentSource(container_url="<SAS-url-to-invoices-container>")
            ),
            "contract": ClassifierDocumentTypeDetails(
                azure_blob_source=AzureBlobContentSource(container_url="<SAS-url-to-contracts-container>")
            ),
        },
    )
)
classifier = poller.result()

You need at least five sample documents per category to train a classifier at all, and more than that for anything you'd trust in production. Once it's built, classifying an incoming document is a single call:

with open("unknown.pdf", "rb") as f:
    poller = client.begin_classify_document("support-doc-classifier", AnalyzeDocumentRequest(bytes_source=f.read()))
result = poller.result()
for doc in result.documents:
    print(f"Classified as: {doc.doc_type} (confidence: {doc.confidence:.2f})")

Step 5: build a custom extraction model for your own document type

When a document type isn't invoices, receipts, or any of the other prebuilt shapes, train your own. This needs a set of labeled training documents in Blob Storage, produced through the labeling tool in Foundry's document intelligence studio or programmatically.

from azure.ai.documentintelligence.models import (
    BuildDocumentModelRequest,
    AzureBlobContentSource,
    DocumentBuildMode,
)

poller = admin_client.begin_build_document_model(
    BuildDocumentModelRequest(
        model_id="acme-service-agreement-v1",
        build_mode=DocumentBuildMode.TEMPLATE,
        azure_blob_source=AzureBlobContentSource(container_url="<SAS-url-to-training-container>"),
        description="Extraction model for Acme's standard service agreement template.",
    )
)
model = poller.result()

Two build modes matter here, and they're not interchangeable. TEMPLATE mode is faster to train and works well when your documents follow a consistent visual layout, the same form filled out differently each time. NEURAL mode handles structural variation better, different layouts that still represent the same document type, at the cost of needing more training examples and longer build time. Start with TEMPLATE unless your documents genuinely vary in structure, not just content.

One naming constraint worth knowing before you hit it: a custom model ID can't start with prebuilt-, since that prefix is reserved for Microsoft's own models across every resource.

Where this fits in the bigger picture

This is the detail that trips people up once they've also worked with the Foundry SDK or Agent Framework elsewhere in this series: Document Intelligence doesn't go through your Foundry project endpoint at all. It has its own resource, its own endpoint (resource.cognitiveservices.azure.com), and its own authentication scope. That's what "Foundry Tools SDK" actually means as a category, prebuilt AI services with tool-specific endpoints, distinct from the Foundry SDK's unified project endpoint that Agent Framework and the Responses API build on.

The practical upshot is the pipeline most teams actually want: run prebuilt-layout over incoming documents, get markdown back, and hand that markdown to a Foundry IQ Knowledge Base as a File Knowledge Source. Document Intelligence handles turning the PDF into clean, structured text. Foundry IQ handles chunking, embedding, and retrieval on top of it. Neither service needs to know the other exists, they just happen to compose well because markdown is a reasonable interchange format for both.

Production considerations before you commit

Don't trust a field just because it came back. A field with a confidence score of 0.41 should not silently flow into a downstream system as if it were as reliable as one scored 0.98. Set a threshold, route low-confidence extractions to human review, and log the confidence distribution over time so a model quietly degrading on a document template change doesn't go unnoticed.
Classifier training minimums are a floor, not a target. Five documents per category is what the service requires to build at all. It is not enough to trust a classifier's accuracy in production. Budget for real evaluation data, held out from training, before routing real documents based on classifier output.
TEMPLATE vs NEURAL is a real tradeoff, not a default to leave unexamined. Picking NEURAL by default because it sounds more capable means slower training and a higher training-data bar for a benefit you may not need if your documents are already visually consistent.
Preview API versions and regional availability move independently of the SDK version. A given SDK release doesn't guarantee every feature is available in every region. Check current regional availability for newer capabilities (certain add-ons, newer prebuilt models) before designing around them.
Markdown output is currently scoped to prebuilt-layout. Don't assume other prebuilt or custom models will hand back the same content format, check per-model support before building a pipeline that assumes markdown everywhere.
Cost scales with pages and capability, not just call count. Add-on features like high-resolution mode and custom model training both carry their own cost beyond the base per-page analysis price. Model this before committing to a design that turns on every add-on by default.

Where this leaves you

The Document Intelligence SDK is easy to undersell because the interesting part of most AI applications feels like it's happening somewhere else, in the model, in the retrieval layer, in the agent's reasoning. But the quality ceiling of everything downstream is set right here, at the point where a physical or scanned document either does or doesn't become text a model can actually use well. Layout extraction to markdown, confidence-aware field extraction, classifiers for mixed intake, and custom models for your own document shapes cover the large majority of real document-processing needs, and all four are a few lines of SDK code once you know which one you need. The judgment call was never really about the API. It's about matching the right one of these four tools to what's actually in your inbound documents.

References

Microsoft. "azure-ai-documentintelligence README." Azure SDK for Python. github.com/Azure/azure-sdk-for-python/blob/main/sdk/documentintelligence/azure-ai-documentintelligence/README.md
Microsoft Learn. "Document Intelligence layout model." learn.microsoft.com/en-us/azure/ai-services/document-intelligence/prebuilt/layout
Microsoft. "Migration guide, azure-ai-documentintelligence." Azure SDK for Python. github.com/Azure/azure-sdk-for-python/blob/main/sdk/documentintelligence/azure-ai-documentintelligence/MIGRATION_GUIDE.md
Microsoft Learn. "Get started with Microsoft Foundry SDKs and endpoints." learn.microsoft.com/en-us/azure/foundry/how-to/develop/sdk-overview
Microsoft Learn. "What is Foundry IQ?" learn.microsoft.com/en-us/azure/foundry/agents/concepts/what-is-foundry-iq

A Senior Engineer's Guide to Foundry IQ, MCP, and the OpenAI Agents SDK

Jubin Soni — Tue, 08 Sep 2026 06:43:36 +0000

Most Foundry writeups assume you're all in on Microsoft's stack end to end, the Agent Framework for orchestration, the Foundry Agent Service for hosting, the Responses API wrapped in Microsoft's own client. That's a reasonable default, but it's not the only shape this can take. Microsoft Foundry hosts OpenAI's own models behind an OpenAI-compatible endpoint, and Foundry IQ exposes every Knowledge Base as a plain MCP server. Put those two facts together and you get a genuinely different setup: OpenAI's own Agents SDK, unmodified, orchestrating a model that happens to be running on Foundry, grounded by a Knowledge Base that happens to be Foundry IQ, with MCP as the only thing that has to agree between them.

This is a hands-on guide to building exactly that. Not because you should always prefer OpenAI's SDK over Microsoft's own tooling, but because knowing this path exists changes how you think about lock-in. If your orchestration layer is a thin, protocol-based client, swapping the model host or the knowledge layer underneath it is a config change, not a rewrite.

The mental model first

Three pieces, from three different places, held together by two protocols:

An OpenAI model, hosted on Microsoft Foundry. Foundry deploys OpenAI's models behind an endpoint that speaks the same wire format as OpenAI's own API, including the Responses API. Point any OpenAI-compatible client at that endpoint with a different base_url and it has no idea it's not talking to OpenAI directly.
A Foundry IQ Knowledge Base, doing the same job it always does: chunking, embedding, indexing, and agentic retrieval over your sources. What matters here is that every Knowledge Base speaks MCP natively. It doesn't care what called it.
The OpenAI Agents SDK, running as your orchestration layer, in your own process, not inside Foundry at all. It calls the Foundry-hosted model for reasoning and generation, and calls the Foundry IQ Knowledge Base as an MCP tool for grounding. Neither call requires Microsoft-specific code.

The thing worth sitting with here: nothing about this setup is a workaround or an unsupported hack. Foundry explicitly documents the OpenAI SDK as the recommended client when you want maximum OpenAI compatibility or the lowest latency path to a Foundry-hosted model. Foundry IQ explicitly exposes MCP as a first-class interface, not an afterthought. This guide is just connecting two things that were each already built to be connected this way.

Prerequisites

You'll need:

A Microsoft Foundry project with an OpenAI model deployed (a gpt-5.1 or similar deployment, created through the Foundry portal or the Foundry SDK).
A Foundry IQ Knowledge Base already built and populated. If you haven't done this before, the short version is a Knowledge Source pointed at your data plus a Knowledge Base wrapping it, both created through azure-search-documents. The full walkthrough, chunking strategy, semantic configuration, and all, is worth its own read if you're starting from zero.
Python 3.10+ with the OpenAI SDK and the Agents SDK installed.

pip install openai openai-agents azure-identity

Step 1: point a plain OpenAI client at your Foundry deployment

Before bringing the Agents SDK into it, confirm the basic connection works with the plain OpenAI client. This is the part that trips people up the least, but it's worth isolating as its own step, because if it doesn't work here, nothing built on top of it will either.

from openai import OpenAI
from azure.identity import DefaultAzureCredential, get_bearer_token_provider

token_provider = get_bearer_token_provider(
    DefaultAzureCredential(), "https://ai.azure.com/.default"
)

client = OpenAI(
    base_url="https://YOUR-RESOURCE-NAME.services.ai.azure.com/api/projects/YOUR-PROJECT/openai",
    api_key=token_provider,
)

response = client.responses.create(
    model="gpt-5.1",
    input="Say hello in one sentence.",
)
print(response.output_text)

Two things to get right here. The base_url is your Foundry project endpoint with /openai on the end, not the raw resource endpoint, and not the older /openai/v1/ Azure OpenAI-specific path (that one still works for Azure OpenAI resources, but the project endpoint is the current recommended shape for Foundry). And api_key accepts a callable token provider, not just a string, which is how Entra ID authentication slots in without you having to manually refresh anything.

Step 2: swap in the token provider and hand the client to the Agents SDK

The Agents SDK doesn't have its own concept of Azure authentication. It just needs an AsyncOpenAI client, and it doesn't care where that client points.

from openai import AsyncOpenAI
from agents import set_default_openai_client, set_tracing_disabled

async_client = AsyncOpenAI(
    base_url="https://YOUR-RESOURCE-NAME.services.ai.azure.com/api/projects/YOUR-PROJECT/openai",
    api_key=token_provider,
)
set_default_openai_client(async_client)

One gotcha worth flagging immediately: the Agents SDK ships with built-in tracing that exports run traces to OpenAI's own platform dashboard by default. That's a sensible default when you're calling OpenAI directly, but it's an odd one once your model calls are routed through Foundry instead, since your traces would still be leaving through a separate, OpenAI-direct path that doesn't share your Foundry project's auth or data boundary. If that matters for your compliance posture, disable it or point it at your own collector:

set_tracing_disabled(True)
# or, to keep tracing but redirect it, register a custom trace processor instead

This is easy to miss because nothing breaks if you leave it on. It just quietly sends run metadata somewhere your Foundry-hosted setup otherwise never touches.

Step 3: connect to the Knowledge Base over MCP

Every Foundry IQ Knowledge Base exposes itself at a predictable MCP endpoint. The Agents SDK's MCPServerStreamableHttp class is built for exactly this kind of self-managed, HTTP-based MCP server.

from agents.mcp import MCPServerStreamableHttp, create_static_tool_filter

kb_server = MCPServerStreamableHttp(
    name="foundry-iq-kb",
    params={
        "url": "https://YOUR-SEARCH-SERVICE.search.windows.net/knowledgebases/team-kb/mcp?api-version=2026-05-01-preview",
        "headers": {"api-key": "YOUR-SEARCH-ADMIN-KEY"},
        "timeout": 15,
    },
    cache_tools_list=True,
    tool_filter=create_static_tool_filter(allowed_tool_names=["knowledge_base_retrieve"]),
)

cache_tools_list=True is worth defaulting to here. A Knowledge Base publishes exactly one tool, knowledge_base_retrieve, and that isn't going to change between requests, so there's no reason to pay a tools/list round trip on every single agent turn. The tool_filter is mostly redundant given there's only one tool to begin with, but it's cheap insurance if the Knowledge Base ever grows a second tool you don't want this particular agent touching.

Step 4: build the agent and run it

With the client and the MCP server both wired up, the agent itself is short.

import asyncio
from agents import Agent, Runner

async def main():
    async with kb_server as server:
        agent = Agent(
            name="support-agent",
            instructions=(
                "Answer questions using the knowledge_base_retrieve tool. "
                "Always call it before answering. Preserve [ref_id:N] citations "
                "from the tool's response in your final answer."
            ),
            model="gpt-5.1",
            mcp_servers=[server],
        )
        result = await Runner.run(agent, "What's our current rate limit on the export API?")
        print(result.final_output)

asyncio.run(main())

The model string here is the Foundry deployment name, not an OpenAI model ID, since every call now routes through the client you registered in Step 2. If you want to stream the response instead of waiting for the full turn, Runner.run_streamed gives you the same event-based streaming interface regardless of which backend is actually generating the tokens:

result = Runner.run_streamed(agent, "What's our current rate limit on the export API?")
async for event in result.stream_events():
    if event.type == "raw_response_event" and hasattr(event.data, "delta"):
        print(event.data.delta, end="", flush=True)

Nothing in either of these two blocks is Foundry-specific or Azure-specific. That's the point. The vendor-specific work all happened in Steps 1 through 3, in the client and connection setup, not in how you define or run the agent.

Where the credentials actually live

Two separate credentials are doing two separate jobs here, and it's worth being precise about which is which, because they fail differently.

The model credential is whatever you passed as api_key on the AsyncOpenAI client, a Foundry project token from DefaultAzureCredential, or a static API key if you're using key-based auth on the resource. This is checked on every responses.create() call the Agents SDK makes internally when the agent reasons or generates a final answer. Scope this to the project, not the whole Foundry resource, using the same RBAC roles you'd use for any other Foundry SDK client (Cognitive Services User is usually sufficient for inference-only access).

The Knowledge Base credential is the api-key header on the MCP server's params, and it's checked independently by Azure AI Search when the knowledge_base_retrieve tool gets called. These two credentials can be, and generally should be, scoped to completely different principals. A key that can call your Foundry model deployment shouldn't automatically be able to query every Knowledge Base on your Search service, and the reverse is just as true. If you're building anything past a prototype, put each behind its own least-privilege identity rather than reusing one Foundry project's admin key for both.

If your Knowledge Base sits over permission-sensitive content, this is also where the on-behalf-of pattern from Foundry IQ's own permission model applies unchanged: thread the requesting user's token through as an additional header on the MCP params, since the KB's enforcement of ingestionPermissionOptions doesn't know or care that the caller this time is the OpenAI Agents SDK instead of the Foundry Agent Service.

Production considerations before you commit

Decide on tracing deliberately, not by default. Leaving the Agents SDK's tracing on means run metadata leaves through an OpenAI-direct path that bypasses your Foundry project's boundary entirely. Turn it off or replace it with a custom processor as a first-day decision, not something you notice in a security review months later.
Cache the tool list, but know when to invalidate it. cache_tools_list=True avoids a redundant round trip, but if you ever change what a Knowledge Base exposes, which is rare but not impossible as Foundry IQ's MCP surface evolves, a long-lived process holding a stale cached tool list will keep calling the old shape until it's restarted.
Separate the model deployment's quota from the Knowledge Base's query load. These are billed and throttled independently. A burst of retrieval-heavy queries against the Knowledge Base won't show up as pressure on your model deployment's tokens-per-minute limit, and the reverse is also true, so alert on both rather than assuming one is a proxy for the other.
Pin the MCP API version. The api-version=2026-05-01-preview query parameter on the Knowledge Base's MCP URL is still a preview surface as of this writing. Track it the same way you'd track any other preview dependency, and don't assume a bare /mcp URL without a version pin will behave identically across a Foundry IQ update.
Keep the instructions honest about tool use. The Agents SDK does not force a tool call. If your instructions say "always call knowledge_base_retrieve" but the model decides a question doesn't need it, you'll get an ungrounded answer with no error. Log whether the tool was actually invoked on each run, not just what the final answer said, if grounding is a correctness requirement rather than a nice-to-have.

Where this leaves you

The interesting thing this setup demonstrates isn't that OpenAI's SDK can technically reach a Foundry endpoint. It's that both vendors built their integration points, an OpenAI-compatible inference endpoint on one side, an MCP-native Knowledge Base on the other, generally enough that they compose without either one knowing the other exists. That's a genuinely different bet than the usual platform story, where the value proposition is staying inside one vendor's tooling end to end. If you're already committed to the OpenAI Agents SDK for orchestration, whether for its tracing, its handoff model, or just team familiarity, you don't have to give that up to use Foundry-hosted models or Foundry IQ's retrieval layer. The protocol boundary is the only thing that has to hold, and both sides already built to it.

References

Microsoft Learn. "Get started with Microsoft Foundry SDKs and endpoints." learn.microsoft.com/en-us/azure/foundry/how-to/develop/sdk-overview
Microsoft Learn. "Use the Azure OpenAI Responses API." Microsoft Foundry. learn.microsoft.com/en-us/azure/foundry/openai/how-to/responses
Microsoft Learn. "How to migrate from Azure AI Inference SDK to OpenAI SDK." Microsoft Foundry. learn.microsoft.com/en-us/azure/foundry/how-to/model-inference-to-openai-migration
OpenAI. "Configuration." OpenAI Agents SDK documentation. openai.github.io/openai-agents-python/config
Microsoft Learn. "What is Foundry IQ?" learn.microsoft.com/en-us/azure/foundry/agents/concepts/what-is-foundry-iq

Foundry IQ Auth, Explained: Managed Identity, OBO, and Everything Between

Jubin Soni — Mon, 31 Aug 2026 23:37:57 +0000

Foundry IQ's pitch is that your agents get one endpoint for grounded, cited, multi-source context instead of a hand-rolled retrieval stack. What doesn't get talked about enough is that "one endpoint" is a bit of a simplification. Underneath it, you're actually managing four separate auth surfaces that don't share a security model: the control plane you use to provision resources, the per-Knowledge-Source credentials that vary by source kind, the connection auth between a Knowledge Base and the agent that calls it, and, the one people get wrong most often, whether the content itself respects the permissions of the person asking.

Get any one of these wrong and you usually don't get an error. You get an agent that confidently answers questions using data the requesting user was never supposed to see. This is a hands-on guide to standing up a real Foundry IQ deployment, Knowledge Sources, a Knowledge Base, and both a Foundry Agent and a Microsoft Agent Framework agent grounded against it, built around getting each of those four auth surfaces right instead of defaulting to the admin key that makes the quickstart work.

The mental model first

Three primitives matter here:

Knowledge Source (KS), a connection to one place your data lives. Could be an existing Azure AI Search index you already own, a file you upload directly, a live Blob container, SharePoint, OneLake, a Fabric data agent, or, notably, an arbitrary external MCP server. Each KS has a kind, and the kind determines its auth model and ingestion pipeline.
Knowledge Base (KB), a named, reusable object that wraps one or more Knowledge Sources plus a chat model. The KB is what your agents actually talk to. Multiple agents can share one KB.
Agentic retrieval engine, the thing that makes this more than "call several sources and concatenate." Given a user turn, it decomposes the query into subqueries, plans which KS each subquery should hit, runs them in parallel, reranks with the semantic ranker, and, if you ask it to, synthesizes one cited natural-language answer instead of dumping raw chunks back at you.

Architecturally, indexed sources (Search Index, Blob, OneLake, SharePoint indexed, SQL) get a managed indexing pipeline. Foundry IQ chunks, embeds, and indexes the content for you. Federated sources (Web, remote SharePoint, MCP servers) are queried live at retrieval time, so no copy of the data sits in your index. That distinction matters a lot for cost, freshness, and latency, and it's the first architectural decision you'll make for each data source.

One thing worth being precise about, because "GA" gets thrown around loosely in this space: as of the current preview cycle, Knowledge Bases and the core indexed sources (Search Index, Azure Blob, OneLake, Web) are generally available on the stable REST API surface. Answer synthesis, configurable reasoning effort, document-level permissions, and multi-turn retrieve are still preview-only capabilities on the newer API version. Check current GA-vs-preview status before you commit a production dependency to a specific feature. This space is moving fast.

The four auth surfaces, up front

Before touching any code, it's worth naming these explicitly, because the hands-on steps below each map to one of them.

Surface	Question it answers	Default	Production choice
Control plane	Who can create or modify KS, KBs, indexes?	`AzureKeyCredential` (admin key)	`DefaultAzureCredential` + RBAC roles
Knowledge Source auth	How does this specific source authenticate to its backing system?	Varies by `kind`, same-service key, upload key, or none	Depends on source. Federated sources often need a dedicated connection
KB to Agent connection	How does the agent calling the KB authenticate to Search?	Shared admin key on the MCP endpoint	`ProjectManagedIdentity` connection, scoped per agent
Content-level permissions	Does the answer respect what the requesting user is allowed to see?	Nothing. Retrieval ignores identity unless you wire it up	ACL and sensitivity-label ingestion plus an OBO token threaded through every call

The first three are about who can operate the system. The fourth is about whether the system, once operating correctly, still leaks data across a permission boundary. Most Foundry IQ writeups stop at the third. The fourth is where real incidents happen, so it gets its own section below, after the build steps.

Prerequisites

You need two things provisioned before you write any code:

An Azure AI Search service (any supported region).
A Microsoft Foundry project with a chat model deployment (for example gpt-4.1-mini or gpt-4o) and an embedding model deployment (for example text-embedding-3-large).

Everything below assumes Python with the preview azure-search-documents SDK, which is the first SDK version to expose the Knowledge Base and Knowledge Source surface, alongside azure-ai-projects for the Foundry Agent bindings.

pip install "azure-search-documents==12.1.0b1" \
            "azure-ai-projects==2.1.0" \
            "azure-identity>=1.19.0" \
            "agent-framework-core>=0.1.0" \
            "agent-framework-openai>=0.1.0"

Set up your clients once and reuse them everywhere:

from azure.core.credentials import AzureKeyCredential
from azure.search.documents.indexes import SearchIndexClient

SEARCH_ENDPOINT = "https://<your-search-service>.search.windows.net"
SEARCH_API_KEY = "<admin-key>"
AOAI_ENDPOINT = "https://<your-foundry-resource>.openai.azure.com"

credential = AzureKeyCredential(SEARCH_API_KEY)
index_client = SearchIndexClient(endpoint=SEARCH_ENDPOINT, credential=credential)

Notice the credential type. AzureKeyCredential gets you moving fast, but for production you'll want to swap this for DefaultAzureCredential and lean on RBAC (Search Service Contributor, Search Index Data Contributor, Cognitive Services User) instead of a static admin key sitting in an env var. Keep that migration on your list from day one. It's a trivial swap later but a real security debt if you skip it.

Step 1: give a Search Index Knowledge Source something to point at

If you already have a vector index, you can wrap it directly, no re-ingestion needed. But to see the whole pipeline, it helps to stand up a small, production-shaped index: a text field, a vector field on an HNSW profile with an Azure OpenAI vectorizer (so query-time embedding happens server-side, not in your app code), and a semantic configuration, which is a hard requirement since the KB's planner leans on the semantic ranker to rerank candidates before synthesis.

from azure.search.documents.indexes.models import (
    SearchIndex, SimpleField, SearchField, SearchFieldDataType,
    VectorSearch, VectorSearchProfile, HnswAlgorithmConfiguration,
    AzureOpenAIVectorizer, AzureOpenAIVectorizerParameters,
    SemanticSearch, SemanticConfiguration, SemanticPrioritizedFields, SemanticField,
)

vectorizer_params = AzureOpenAIVectorizerParameters(
    resource_url=AOAI_ENDPOINT,
    deployment_name="text-embedding-3-large",
    api_key="<aoai-key>",
    model_name="text-embedding-3-large",
)

index = SearchIndex(
    name="docs-index",
    fields=[
        SimpleField(name="id", type=SearchFieldDataType.String, key=True),
        SearchField(name="chunk", type=SearchFieldDataType.String),
        SearchField(
            name="chunk_vector",
            type=SearchFieldDataType.Collection(SearchFieldDataType.Single),
            vector_search_dimensions=3072,
            vector_search_profile_name="hnsw",
        ),
    ],
    vector_search=VectorSearch(
        profiles=[VectorSearchProfile(name="hnsw", algorithm_configuration_name="alg", vectorizer_name="aoai")],
        algorithms=[HnswAlgorithmConfiguration(name="alg")],
        vectorizers=[AzureOpenAIVectorizer(vectorizer_name="aoai", parameters=vectorizer_params)],
    ),
    semantic_search=SemanticSearch(
        default_configuration_name="semantic",
        configurations=[SemanticConfiguration(
            name="semantic",
            prioritized_fields=SemanticPrioritizedFields(content_fields=[SemanticField(field_name="chunk")]),
        )],
    ),
)
index_client.create_or_update_index(index)

Once the index has documents, wrap it in a SearchIndexKnowledgeSource:

from azure.search.documents.indexes.models import SearchIndexKnowledgeSource, SearchIndexKnowledgeSourceParameters

ks_index = SearchIndexKnowledgeSource(
    name="ks-docs-index",
    description="Existing product docs index.",
    search_index_parameters=SearchIndexKnowledgeSourceParameters(
        search_index_name="docs-index",
        semantic_configuration_name="semantic",
        source_data_fields=[{"name": "id"}, {"name": "chunk"}],
    ),
)
index_client.create_or_update_knowledge_source(ks_index)

One gotcha worth flagging here: the baseFilter you can set on a Search Index KS narrows the queryable subset for every KB that references it, and its semantics are different from a per-call filterAddOn at retrieve time. Don't conflate the two. One is a source-level ceiling, the other is a request-level refinement.

Step 2: upload a file directly, no storage account required

For content that doesn't live in a managed data store, the File KS lets you POST documents straight into Foundry IQ and it handles chunking and embedding for you.

from azure.search.documents.indexes.models import FileKnowledgeSource, FileKnowledgeSourceParameters, KnowledgeSourceIngestionParameters, KnowledgeSourceAzureOpenAIVectorizer

ks_file = FileKnowledgeSource(
    name="ks-uploaded-pdf",
    description="Directly uploaded reference PDF.",
    file_parameters=FileKnowledgeSourceParameters(
        ingestion_parameters=KnowledgeSourceIngestionParameters(
            content_extraction_mode="minimal",
            embedding_model=KnowledgeSourceAzureOpenAIVectorizer(azure_open_ai_parameters=vectorizer_params),
        ),
    ),
)
index_client.create_or_update_knowledge_source(ks_file)

with open("reference.pdf", "rb") as fh:
    uploaded = index_client.upload_knowledge_source_file("ks-uploaded-pdf", fh.read(), filename="reference.pdf")

A gotcha here too: the file upload endpoint is a plain REST route, not the usual OData-style ('name') addressing pattern the rest of the Search API uses. A JSON-only middleware in front rejects binary bodies sent the OData way. If you're calling this from something other than the typed SDK, get the URL shape right or you'll spend longer than you'd like debugging a 415. Also, embedding happens synchronously after upload, so poll the KS status until synchronizationStatus reports active before you query the KB. Querying too early raises a validation error rather than silently returning partial results.

Step 3: federate a live external source over MCP

This is the part that's genuinely new relative to a typical RAG stack. A Knowledge Source can be another MCP server entirely. No copying, no indexing pipeline, every retrieve does a live tools/call against the upstream server.

from azure.search.documents.indexes.models import McpServerKnowledgeSource, McpServerKnowledgeSourceParameters

ks_mcp = McpServerKnowledgeSource(
    name="ks-external-mcp",
    description="Federated live source over MCP.",
    mcp_server_parameters=McpServerKnowledgeSourceParameters(
        server_url="https://learn.microsoft.com/api/mcp",
        tools=[{
            "name": "microsoft_docs_search",
            "outputParsing": {"kind": "auto"},
            "inclusionMode": "reranked",
            "maxOutputTokens": 4096,
        }],
    ),
)
index_client.create_or_update_knowledge_source(ks_mcp)

The tools[].name you specify has to exactly match a tool name the upstream MCP server actually publishes. There's no fuzzy matching. If outputParsing.kind: "auto" fails to infer the right shape for a given server's responses, fall back to explicit "text" or "structured". For servers that require an API key, the canonical pattern is a Foundry CustomKeys connection rather than embedding secrets in the KS definition.

Step 4: assemble the Knowledge Base

The KB is where you decide how hard the planner should work and what shape you want the output in.

from azure.search.documents.indexes.models import KnowledgeBase, KnowledgeBaseAzureOpenAIModel, KnowledgeSourceReference
from azure.search.documents.knowledgebases.models import KnowledgeRetrievalLowReasoningEffort, KnowledgeRetrievalOutputMode

gpt_params = AzureOpenAIVectorizerParameters(
    resource_url=AOAI_ENDPOINT, deployment_name="gpt-4.1-mini",
    api_key="<aoai-key>", model_name="gpt-4.1-mini",
)

kb = KnowledgeBase(
    name="team-kb",
    description="Unified KB over indexed docs, an uploaded file, and a federated MCP source.",
    models=[KnowledgeBaseAzureOpenAIModel(azure_open_ai_parameters=gpt_params)],
    knowledge_sources=[
        KnowledgeSourceReference(name="ks-docs-index"),
        KnowledgeSourceReference(name="ks-uploaded-pdf"),
        KnowledgeSourceReference(name="ks-external-mcp"),
    ],
    retrieval_reasoning_effort=KnowledgeRetrievalLowReasoningEffort(),
    output_mode=KnowledgeRetrievalOutputMode.ANSWER_SYNTHESIS,
    answer_instructions="Answer only from retrieved content. Preserve [ref_id:N] citations.",
)
index_client.create_or_update_knowledge_base(kb)

Two settings are worth deliberating over rather than defaulting blindly:

retrieval_reasoning_effort (minimal, low, or medium) trades latency and token spend against how thoroughly the planner decomposes and iterates on a query. low is a reasonable production default. Reach for medium only when you're seeing the planner short-circuit on genuinely multi-part questions.
output_mode: extractiveData hands you raw ranked chunks, useful if you want to do your own synthesis or need maximum auditability. answerSynthesis gives you a single cited natural-language answer, which is what most agent integrations actually want.

Step 5: query it and watch it actually plan

The payoff case is a query that no single source can answer alone.

from azure.search.documents.knowledgebases import KnowledgeBaseRetrievalClient
from azure.search.documents.knowledgebases.models import KnowledgeBaseRetrievalRequest, KnowledgeBaseMessage, KnowledgeBaseMessageTextContent

retrieval_client = KnowledgeBaseRetrievalClient(endpoint=SEARCH_ENDPOINT, credential=credential, knowledge_base_name="team-kb")

request = KnowledgeBaseRetrievalRequest(
    messages=[KnowledgeBaseMessage(role="user", content=[KnowledgeBaseMessageTextContent(
        text="Compare what our docs say about rate limits with what Microsoft Learn says about retrieval throttling."
    )])],
    include_activity=True,
)
result = retrieval_client.retrieve(request)

Set include_activity=True and you get the planner's trace back: which subqueries it generated, which KS each one hit, and how results were reranked before synthesis. This is worth logging in any non-trivial deployment. It's your debugging surface when the answer looks wrong and you need to know whether the planner queried the wrong source, or queried the right source and got a bad rerank.

Retrieval is also conversation-aware. Append the prior assistant turn and a follow-up user message to messages, and the planner uses that context to scope its next round of subqueries, genuinely useful for the "wait, tell me more about X from that answer" pattern that trips up a lot of naive single-shot RAG.

Step 6: every KB is already an MCP server, use it directly

You don't have to go through the SDK at all. Every Knowledge Base exposes itself at:

{SEARCH_ENDPOINT}/knowledgebases/{KB_NAME}/mcp?api-version=2026-05-01-preview

as a standard JSON-RPC 2.0 MCP endpoint (JSON or SSE-streamable), with exactly one tool published, knowledge_base_retrieve. Auth is the same Search admin key on the api-key header, or, for federated sources that need user identity such as Remote SharePoint or WorkIQ, an x-ms-query-source-authorization header carrying an on-behalf-of user token. That means Claude Desktop, VS Code Copilot, a custom script, or any other MCP-capable client can hit the exact same retrieval pipeline your agents use, with zero custom integration code. This is arguably the most senior-engineer-relevant design decision in the whole product. The KB isn't a bespoke API you have to wrap, it's already speaking a protocol your tooling ecosystem understands.

Step 7: ground a Foundry Agent

Wiring a KB into the Foundry Agent Service is a three-step dance. Create a RemoteTool project connection pointing at the KB's MCP URL, create an agent declaring an mcp tool against that connection with knowledge_base_retrieve allow-listed, then drive it through the Responses API.

from azure.ai.projects import AIProjectClient
from azure.ai.projects.models import MCPTool, PromptAgentDefinition
from azure.identity import DefaultAzureCredential

project_client = AIProjectClient(endpoint="<foundry-project-endpoint>", credential=DefaultAzureCredential())

mcp_tool = MCPTool(
    server_label="team-kb",
    server_url=f"{SEARCH_ENDPOINT}/knowledgebases/team-kb/mcp?api-version=2026-05-01-preview",
    require_approval="never",
    allowed_tools=["knowledge_base_retrieve"],
    project_connection_id="team-kb-connection",  # created via a RemoteTool connection beforehand
)

agent_def = PromptAgentDefinition(
    model="gpt-4.1-mini",
    instructions="Always call knowledge_base_retrieve before answering. Preserve [ref_id:N] citations.",
    tools=[mcp_tool],
)
project_client.agents.create_version(agent_name="support-agent", definition=agent_def)

openai_client = project_client.get_openai_client()
conversation = openai_client.conversations.create()
response = openai_client.responses.create(
    conversation=conversation.id,
    input="What's our current rate limit policy?",
    extra_body={"agent_reference": {"name": "support-agent", "type": "agent_reference"}},
)

Use authType=ProjectManagedIdentity on the connection so the project authenticates to Search as itself, rather than storing a per-user token on a shared connection object. That covers this auth surface, whether the agent can call the KB at all. It says nothing about whether the content it retrieves respects the permissions of whoever's talking to the agent. That's a separate mechanism, covered in full below.

Step 8: or plug it into Microsoft Agent Framework instead

If you're not using Foundry Agent Service, the Agent Framework's MCPStreamableHTTPTool connects to the same endpoint directly:

from agent_framework import Agent, MCPStreamableHTTPTool
from agent_framework_openai import OpenAIChatClient
import httpx

mcp_http = httpx.AsyncClient(headers={"api-key": SEARCH_API_KEY, "Accept": "application/json, text/event-stream"})

async with MCPStreamableHTTPTool(
    name="team_kb",
    url=f"{SEARCH_ENDPOINT}/knowledgebases/team-kb/mcp?api-version=2026-05-01-preview",
    http_client=mcp_http,
    allowed_tools=["knowledge_base_retrieve"],
    load_prompts=False,  # the KB MCP server is stateless, don't try prompts/list
) as kb_tool:
    agent = Agent(
        client=OpenAIChatClient(azure_endpoint=AOAI_ENDPOINT, api_key="<key>", api_version="preview", model="gpt-4.1-mini"),
        name="SupportAgent",
        instructions="Call team_kb-knowledge_base_retrieve before answering. Preserve citations.",
        tools=kb_tool,
    )
    response = await agent.run("What's our current rate limit policy?")

load_prompts=False isn't optional decoration. The KB's MCP server doesn't implement prompts/list, and frameworks that call it eagerly on connect will error out if you don't disable it explicitly.

Surface 4: making retrieval actually respect who's asking

This is the part worth slowing down for, because it's the one gap between "the demo works" and "this is safe to point at real enterprise content."

By default, a Knowledge Base retrieves without any notion of the requesting user. If your Search Index KS was built over a SharePoint site with document-level permissions, and you don't do anything further, Foundry IQ will happily surface a chunk from a document the asking user has no access to. The vector index doesn't know who's asking, and the KB won't stop it for you. Getting this right is two separate steps, and skipping either one leaves the gap open.

Step A, get permission metadata into the index at ingestion time. For indexed sources that support full ACLs, ADLS Gen2 and SharePoint (indexed), set ingestionPermissionOptions to include the identity fields you need:

search_index_parameters = SearchIndexKnowledgeSourceParameters(
    search_index_name="docs-index",
    semantic_configuration_name="semantic",
    # pulls group/user ACLs and sensitivity labels into the index alongside content
    ingestion_permission_options=["groupIds", "userIds", "sensitivityLabel"],
)

Blob and OneLake only carry sensitivityLabel. They don't support full ACL ingestion, so document-level access control on those sources has to be enforced elsewhere in your architecture (container-level access, for instance, or just don't put mixed-sensitivity content in a single Blob-backed KS).

Step B, thread the requesting user's identity through every retrieve call. Indexed metadata sits inert until a request carries an identity to check it against. That identity travels as an on-behalf-of (OBO) token:

request = KnowledgeBaseRetrievalRequest(
    messages=[...],
    include_activity=True,
    # the requesting user's OBO token, checked against ACLs indexed in Step A
    x_ms_query_source_authorization=user_obo_token,
)

Over the KB's MCP endpoint, the same token goes on the x-ms-query-source-authorization header rather than a request field. Either way, it's the piece that turns "we indexed who can see this" into "we actually enforced it for this specific call." If your agent runtime pools requests behind a single service identity and doesn't propagate the calling user's token down to the retrieve call, Steps A and B are both true and enforcement still doesn't happen. The failure mode is silent, not an error.

The same OBO token is what governs federated sources with per-user semantics, Remote SharePoint and WorkIQ, since there's no indexed copy to attach ACLs to in the first place. The upstream system checks the token itself on every live query.

A concrete failure mode worth testing for: stand up a KB over content with mixed permissions, query it as two users with different access, and diff the retrieved chunks and citations. If the low-privilege user ever gets a chunk from a document the high-privilege user's account owns, you've got a Step A or Step B gap. It's worth making this diff test a permanent fixture in CI for any KB backed by access-controlled content, not a one-time manual check.

Surfaces 1 through 3, hardened: a checklist

Control plane. Swap AzureKeyCredential for DefaultAzureCredential and grant Search Service Contributor (provisioning), Search Index Data Contributor (data-plane read and write), and Cognitive Services User (Foundry model access) instead of holding a standing admin key in an env var.
Knowledge Source auth. Same-service KS kinds (Search Index, File) ride the Search service's own credential, nothing extra to manage. Federated sources that need a key, an MCP server behind auth, a data source needing a shared secret, should go through a Foundry CustomKeys connection rather than a literal string in the KS definition, so the secret is rotatable and auditable independently of your KS config.
KB to Agent connection. Use authType=ProjectManagedIdentity on the RemoteTool connection so the agent authenticates to Search as the project's own identity, scoped to that project, not a shared key that every agent in the tenant could also use if they found it. This is the difference between "an agent can call this KB" and "anyone with this string can call every KB on the service."
GA vs. preview. Knowledge Bases and core indexed sources (Search Index, Blob, OneLake, Web) are GA on the stable API. Document-level permissions, answer synthesis, and multi-turn retrieve are still preview-only as of this writing. Don't let a preview-only permissions feature be the only thing standing between your KB and an access-control gap. Pin your API version deliberately and track when it goes GA.
Observability. include_activity=True on every retrieve call gives you the planner's trace, which KS got queried, with what, and that's your primary tool for confirming a permission-scoped query actually stayed scoped, not just for debugging relevance.

Where this leaves you

The interesting engineering bet Foundry IQ is making isn't the managed indexing pipeline. That's table stakes at this point. It's that the Knowledge Base's interface is MCP, not a bespoke SDK surface, and that permission enforcement is a first-class, if opt-in, part of that interface via ACL ingestion and OBO propagation. That combination is also exactly where the risk concentrates. A protocol that's trivially easy for any MCP client to call is only as safe as the identity you remember to attach to every single request. If you're already committed to Azure AI Search and Microsoft Foundry, treat Steps A and B above as non-negotiable for any KB backed by access-controlled content, not an optional hardening pass you get to later. If you're not on Azure, the pattern is worth stealing regardless of platform: index permission metadata alongside content, and make every retrieval call carry the identity it's answering on behalf of.

References

Microsoft Learn. "What is Foundry IQ?" learn.microsoft.com/en-us/azure/foundry/agents/concepts/what-is-foundry-iq
Microsoft Foundry Blog. "Foundry IQ: Build smarter agents faster with unified knowledge and serverless retrieval." devblogs.microsoft.com/foundry/build-smarter-agents-faster-with-foundry-iq
Azure AI Search Team. "Foundry IQ: Unlock knowledge retrieval for agents." Microsoft Tech Community. techcommunity.microsoft.com/blog/azure-ai-foundry-blog/foundry-iq-unlocking-ubiquitous-knowledge-for-agents/4470812
Sunavala, Farzad. "Mastering Foundry IQ." Microsoft Foundry Forgebook. microsoft-foundry.github.io/forgebook/notebook/mastering-foundry-iq
Serra, James. "Making Sense of Microsoft's AI Strategy: Work IQ, Fabric IQ, Foundry IQ." jamesserra.com/archive/2026/02/making-sense-of-microsofts-ai-strategy-work-iq-fabric-iq-foundry-iq

Video and Audio as Knowledge Sources: Content Understanding into Microsoft Foundry IQ

Jubin Soni — Tue, 25 Aug 2026 02:10:07 +0000

Your largest corpus is the one nobody indexed. Here's how to turn recordings into grounding data an agent can cite.

The corpus you already have and never indexed

Every organization is sitting on years of recorded meetings, support calls, training sessions, conference talks, and screen recordings. Almost none of it is retrievable. When someone asks "what did we decide about the vendor migration," the answer exists — in a 47-minute recording nobody will ever scrub through.

The reflex fix is to bolt a transcription service onto your RAG pipeline: run Whisper, dump the text into a blob container, index it. It works, sort of, and then you discover what you lost. The transcript has no speakers, no timestamps you can link back to, no slide content, no distinction between the presenter reading a bullet and someone in the room disagreeing with it. Your agent can now quote the meeting but can't tell you when it happened or who said it.

Azure Content Understanding is Microsoft's answer to that gap: it ingests documents, audio, images, and video and extracts the most critical information to power well-grounded generative and agentic solutions, combining Document Intelligence's traditional AI with LLM-based content reasoning. And as of Build 2026 it is integrated with Foundry IQ standard mode for built-in content extraction inside Microsoft's retrieval and agent workflows.

That integration is the subject of this article. Specifically, the part everyone gets wrong: there are two ways to get media into a knowledge base, they are not equivalent, and the one people assume exists is the one you should verify before you plan a sprint around it.

This is a companion to my earlier piece on connecting a Foundry IQ knowledge base to LangGraph over MCP. That one covered retrieval. This one covers what you feed it.

Architecture

Figure 1 — Two paths into the same knowledge base. The native path is a flag on the knowledge source; the explicit path runs analyzers yourself and lands the output as text.

The two paths, stated plainly

Path A — native extraction inside the ingestion pipeline. Setting the contentExtractionMode property to standard on file-based indexed knowledge sources (Azure Blob, SharePoint, OneLake) enables Content Understanding functionality within the ingestion pipeline. One property. No orchestration code. This shipped in the Foundry IQ 2026-05-01-preview release, which focused on richer Content Understanding extraction and image serving for multimodal agentic retrieval.

Path B — explicit analysis, then index the output. You run Content Understanding yourself, write the resulting Markdown and fields to blob storage, and point a normal knowledge source at that. More moving parts, complete control.

Here is the honest caveat, and I would rather you hear it from me than discover it in week three: Microsoft's Foundry IQ extraction announcements emphasize document understanding — layout, tables, figures, and document-embedded images — rather than audio and video ingestion. Content Understanding standard mode itself is documented for documents, images, audio, and video. Whether your blob knowledge source will accept an .mp4 today, in your region, at your API version, is a question you should answer with a five-minute test rather than an assumption.

So: test Path A first, build on Path B if it doesn't cover your media types. Path B is what this tutorial walks through in detail, because it works regardless, and because understanding it makes Path A trivial to adopt when it covers you.

Prerequisites

A Microsoft Foundry resource, and a Content Understanding resource.
An LLM deployment for analyzers. Analyzers are powered by LLM and embedding models you deploy in Foundry, and GPT-5.2 improves custom field extraction enough to avoid prompt-engineering gymnastics on mixed layouts, domain-specific language, and multilingual content. Analyzers built on GPT-4.1 continue to run unchanged.
An Azure AI Search service for the knowledge base.

pip install 'markitdown[az-content-understanding]' azure-search-documents azure-identity

Pin your API version before writing a line of code

Content Understanding's GA API is 2025-11-01. The preview versions 2024-12-01-preview and 2025-05-01-preview were slated for retirement on July 15, 2026 — a date that has now passed, so if you inherited a codebase targeting either, it is already broken or about to be.

Note the asymmetry with Foundry IQ, which is on 2026-05-01-preview for the content-extraction features. You will be running a GA content service against a preview retrieval service. Plan your support expectations accordingly.

Step 1 — Choose the analyzer, and know what each modality costs you

Content Understanding ships prebuilt analyzers per modality, and MarkItDown auto-selects among them: documents route to prebuilt-documentSearch, video to prebuilt-videoSearch, and audio to prebuilt-audioSearch.

They are not interchangeable, and the differences determine what your agent can cite:

	Documents	Audio	Video
Prebuilt analyzer	`prebuilt-documentSearch`	`prebuilt-audioSearch`	`prebuilt-videoSearch`
Primary output	Layout-aware Markdown	Transcript with structure	Transcript plus visual context
Structure preserved	Headings, tables, figure descriptions	Utterance boundaries	Scene and segment boundaries
Natural citation anchor	Page and figure ID	Timestamp	Timestamp and frame
Grounding you gain	Table cells stay in their table	Who spoke, and when	What was on screen, not just said
What you still lose	Nothing much, this is the mature path	Visual aids referenced verbally	Fine detail in dense slides
Custom field schema	Yes	Yes	Yes
Best for	Contracts, reports, forms	Support calls, interviews	Meetings, demos, training

The row that matters most is natural citation anchor. A document chunk cites a page; an audio chunk cites a timestamp. If you flatten audio into plain text before indexing, you throw away the only anchor that makes a recording navigable — and no amount of clever chunking downstream will recover it.

Step 2 — The fastest path from a recording to indexable text

MarkItDown with the Content Understanding backend is the shortest route, and it is genuinely a few lines. Zero configuration auto-selects the analyzer per file type:

from markitdown import MarkItDown

md = MarkItDown(cu_endpoint="<content_understanding_endpoint>")

doc = md.convert("report.pdf")     # → prebuilt-documentSearch
video = md.convert("meeting.mp4")  # → prebuilt-videoSearch
audio = md.convert("call.wav")     # → prebuilt-audioSearch

print(video.markdown)

The output is Markdown with headings, tables, and figure descriptions inline — exactly the shape downstream chunkers and embedding models prefer. That last clause is the whole argument for this approach: you are not inventing a format, you are producing the format the rest of the stack already wants.

With a custom analyzer, the output carries extracted fields as YAML front matter above the body:

md = MarkItDown(
    cu_endpoint="<content_understanding_endpoint>",
    cu_analyzer_id="my-meeting-analyzer",
)
result = md.convert("standup-2026-08-14.mp4")

---
contentType: video
fields:
  MeetingTitle: Vendor migration review
  Decision: Proceed with phased cutover
  Owner: A. Rivera
---
<!-- 00:04:12 -->
...

That front matter is not decoration. It is your metadata filter, your citation payload, and the difference between "the agent found a relevant meeting" and "the agent told me we decided to proceed, and here is the timestamp."

Step 3 — Design a field schema worth extracting

Prebuilt analyzers give you good transcripts. Custom analyzers give you answers to questions you ask repeatedly — and those are what make retrieval feel like it understands your business rather than your file formats.

Two constraints to plan around:

Custom analyzers are built in Content Understanding Studio, not the Foundry portal. The Foundry portal surfaces prebuilt analyzers and a playground, with a deep link into CU Studio that preserves your project context, but custom analyzer creation lives in Studio. Expect to move between two tools.

Standard mode is per-file. Standard mode handles single files with straightforward field extraction — documents, images, audio, or video without cross-file analysis or complex reasoning. Pro mode is for multi-step reasoning and cross-file analysis. Since Foundry IQ's integration is with standard mode, per-file extraction is your ingestion contract. Questions like "which meetings contradicted the Q2 plan" are retrieval-time work for the agentic retrieval engine, not ingestion-time work for the analyzer.

A schema that earns its keep on meeting recordings:

Field	Why it earns its place
`Decision`	The single most-asked question of any meeting corpus
`Owner`	Turns retrieval into accountability
`DueDate`	Enables recency and deadline filters
`SystemsMentioned`	Lets an agent scope to "anything about the billing service"
`UnresolvedQuestions`	Surfaces what a summary would smooth over

Resist the urge to extract a summary field. The retrieval engine synthesizes at query time against the actual question; a pre-baked summary just adds a lossy paraphrase your agent might cite instead of the source.

One operational note from Microsoft that applies whenever you change models: run side-by-side against your existing eval set before flipping production traffic, since confidence scores, latency, and output accuracy can all shift with a new model.

Step 4 — Land the output where a knowledge source can reach it

Write the Markdown to blob storage, one file per recording, with the front matter intact:

import pathlib
from azure.identity import DefaultAzureCredential
from azure.storage.blob import BlobServiceClient
from markitdown import MarkItDown

md = MarkItDown(cu_endpoint=CU_ENDPOINT, cu_analyzer_id="my-meeting-analyzer")
blobs = BlobServiceClient(
    account_url="https://stfoundryiqprod.blob.core.windows.net",
    credential=DefaultAzureCredential(),
).get_container_client("meeting-transcripts")


def ingest(path: str) -> str:
    """Analyze one recording and land the result as indexable Markdown."""
    result = md.convert(path)
    name = pathlib.Path(path).stem + ".md"
    blobs.upload_blob(name, result.markdown.encode("utf-8"), overwrite=True)
    return name

Keep the source recording and the derived Markdown in separate containers. You want the knowledge source pointed at text only, and you want the original media addressable for playback when a citation resolves. Mixing them means either indexing binaries you can't use or losing the link back to the thing a user actually wants to watch.

Step 5 — Create the knowledge source and the knowledge base

Now it is an ordinary Foundry IQ ingestion. Foundry IQ automates document chunking, vector embedding generation, and metadata extraction for indexed knowledge sources, and schedules recurring indexer runs for incremental refresh.

from azure.identity import DefaultAzureCredential
from azure.search.documents.indexes import SearchIndexClient
from azure.search.documents.indexes.models import (
    AzureBlobKnowledgeSource,
    AzureBlobKnowledgeSourceParameters,
    KnowledgeBase,
    KnowledgeSourceReference,
)

client = SearchIndexClient(endpoint=SEARCH_ENDPOINT,
                           credential=DefaultAzureCredential())

source = AzureBlobKnowledgeSource(
    name="meeting-recordings-ks",
    description=(
        "Transcribed and structured meeting recordings: decisions, owners, "
        "due dates and unresolved questions, with timestamps. "
        "Use for questions about what was decided, by whom, and when."
    ),
    azure_blob_parameters=AzureBlobKnowledgeSourceParameters(
        connection_string=BLOB_CONNECTION,
        container_name="meeting-transcripts",
        # Path A: enable Content Understanding inside the ingestion pipeline.
        # Test this against your actual media types before relying on it.
        content_extraction_mode="standard",
    ),
)
client.create_or_update_knowledge_source(knowledge_source=source)

client.create_or_update_knowledge_base(knowledge_base=KnowledgeBase(
    name="meetings-kb",
    knowledge_sources=[KnowledgeSourceReference(name="meeting-recordings-ks")],
    retrieval_instructions=(
        "Prefer the most recent meeting when decisions conflict. "
        "Always surface the timestamp and speaker with any quoted decision."
    ),
))

Two things to be deliberate about.

The description is load-bearing. The agentic retrieval engine uses it to plan queries and select sources, so write it the way you would brief a new colleague: what is in here, and what kinds of question it answers. "Meeting transcripts" is a wasted field.

The content_extraction_mode="standard" flag is your Path A test. Point it at a container holding one .mp4 and one .md, run the indexer, and look at what got indexed. If the media file produced chunks, you can skip Steps 2 through 4 for that file type. If it didn't, your pipeline is already correct and you have lost five minutes.

What happens at query time

Figure 2 — Ingestion runs once per recording; retrieval runs per question. The timestamp survives both, which is the point.

Nothing about retrieval changes because the source was a video. That is the entire payoff: the agentic retrieval engine plans queries, selects sources, runs parallel searches, and aggregates results, returning extractive data with citations so agents can reason over raw content and trace answers to source documents. Your recordings are now just documents that happen to have timestamps.

The 2026-05-01-preview release also added image serving — surfacing document-embedded images during agentic retrieval. For a slide-heavy deck or a demo recording, that is the difference between "the presenter showed an architecture diagram" and actually returning the frame. Combined with figure extraction from Office files, where each figure is retrievable by ID via GET /contentunderstanding/analyzerResults/{operationId}/files/figures/{figureId}, you can build citations that resolve to a picture rather than a paraphrase of one.

Making citations survive the pipeline

This is where most media-RAG implementations quietly fail. The transcript is indexed, retrieval works, the answer is correct — and the citation says "meeting-2026-08-14.md," which is useless to someone who wants the 12 seconds where the decision was made.

Three rules:

Never strip the front matter during chunking. If your chunker treats YAML as noise, the extracted fields never reach the index and your metadata filters silently match nothing.
Keep timestamp markers inside the chunk text, not only in metadata. Extractive retrieval returns content; if the timestamp lives only in a sidecar field, the model has nothing to quote.
Store a deterministic link back to the media. A blob URL plus a timestamp fragment costs you one field at ingestion and turns every citation into a playable link.

The test: ask your agent a question, take the citation it returns, and try to get to the exact moment in the recording. If you can't do it in one click, the pipeline isn't finished, however good the answer text looks.

Troubleshooting

Symptom	Likely cause	What to do
Media files in the container produce no chunks	The knowledge source didn't ingest that file type natively	Fall back to Path B: analyze with CU, index the Markdown output
`404` or version errors from the CU endpoint	Code still targeting a retired preview API	Move to the GA API version `2025-11-01`
Custom analyzer missing in the Foundry portal	Custom analyzers are created in CU Studio	Use the deep link from Foundry; project context is preserved
Extraction quality dropped after a model change	Confidence scores, latency, and accuracy shift between models	Re-run your eval set side by side before shifting production traffic
Fields extracted but never filterable	Front matter stripped during chunking	Preserve YAML through the chunker; verify fields landed in the index
Citations resolve to a file, not a moment	Timestamps kept only in metadata	Keep markers in chunk text and store a media URL per chunk
Cross-file questions return thin answers	Standard mode is per-file by design	Let the agentic retrieval engine handle cross-document reasoning at query time
Office figures not retrievable	Figures need fetching by ID	Use the `figures/{figureId}` endpoint referenced from the Markdown

Where to go next

A few things landed or were slated for July 2026 that change the calculus here, and are worth confirming against current docs before you design around them: a synchronous API for Read and Layout, an agentic understanding mode for complex documents, data zone and global zone processing for residency, improved custom analyzer training from your own examples, and labeled training data no longer being stored in CU so training inputs stay in your own storage. That last one matters most if compliance review is what's blocking you.

The natural follow-on build is a routing layer: send short call recordings straight through prebuilt-audioSearch, send slide-heavy sessions through prebuilt-videoSearch so you keep the visual channel, and reserve custom analyzers for the recording types you query weekly. Extraction quality and cost both track how well that routing matches your corpus.

References

What is Foundry IQ? — knowledge sources, indexing automation, agentic retrieval and ACL enforcement.
Content Understanding standard and pro modes — per-file vs. cross-file analysis, and the preview API retirement notice.
Content Understanding models and deployments
Foundry vs. Content Understanding Studio — which tool does what.
Supported document formats and input file limits
Build a RAG solution with Content Understanding
Content Understanding in LangChain
What's new in Azure Content Understanding at Build 2026 — the Foundry IQ standard-mode integration, MarkItDown backend, GPT-5.2 analyzers, and the July roadmap.
Improved data processing features in Foundry IQ: richer content extraction and data enrichment — contentExtractionMode, SharePoint indexing expansion, and image serving.
Foundry IQ: improve recall by up to 54% with knowledge bases
Build 2026 Session BRK242 — "Turn your agents into action" — agentic understanding mode and the Foundry IQ integration, demoed end to end.
MarkItDown on GitHub — the CU-backed converter used throughout this article.
Microsoft Agent Framework — for registering CU as an agent tool instead of a pipeline stage.

How to Connect a Foundry IQ Knowledge Base to LangGraph over MCP

Jubin Soni — Tue, 18 Aug 2026 08:43:07 +0000

A step-by-step guide to grounding a LangGraph agent in Microsoft Foundry IQ agentic retrieval — without rebuilding your RAG pipeline.

Why this integration is worth doing

If you build agents on LangGraph and your enterprise content lives in Azure, you have probably written the same code twice: a chunker, an embedding job, a vector store, a retriever, a reranker, and a permissions filter bolted on at the end. Every new agent gets its own copy. Every copy drifts.

Foundry IQ moves that work behind a single endpoint. A knowledge base wraps one or more knowledge sources, and the agentic retrieval engine handles query planning, parallel execution, semantic reranking, and (optionally) answer synthesis. Crucially for anyone outside the Microsoft agent stack: every knowledge base is also a standalone MCP server exposing one tool, knowledge_base_retrieve. Any MCP-compatible client can call it — including LangGraph, via langchain-mcp-adapters.

That is the whole integration. The interesting parts are the four places it does not behave like a normal retriever, which this tutorial covers in detail:

The MCP tool result has a different shape from the REST/SDK retrieve response.
Bearer tokens expire, and a static headers dict will fail an hour into a long-running graph.
Per-user permission filtering requires a second token, distinct from your service credential.
The knowledge base is itself a planner, so you have two planners per turn and need to decide who does what.

By the end you will have a working LangGraph agent grounded in a Foundry IQ knowledge base, with citations preserved in graph state and a token provider that survives long sessions.

What you should already know: LangGraph basics (StateGraph, ToolNode, the ReAct loop), and enough Azure to create a resource and assign a role.

Architecture

Figure 1 — The knowledge base owns retrieval. LangGraph owns orchestration. MCP is the contract between them.

Three things are worth noticing before you write any code.

The knowledge base is reusable. It is not scoped to one agent. The same knowledge base can ground a LangGraph agent, a Foundry Agent Service agent, and a Copilot integration simultaneously. That is the point of the abstraction, and it changes how you name things — name knowledge bases after topics (hr-policy-kb, product-docs-kb), not after the agent that happens to consume them first.

Sources come in two flavours. Indexed sources (Blob, OneLake, an existing search index) are ingested, chunked and vectorized into an index on your search service. Federated sources (remote SharePoint, web, MCP servers) are queried live at retrieval time and never ingested. This distinction matters for permissions, which we return to in Step 7.

Identity flows in two channels. Your application authenticates to the search service with a service identity. Optionally, you also pass the end user's identity in a separate header so the engine filters documents that user may not see. Conflating these two is the most common source of "why is everyone seeing everything" bugs.

Prerequisites

An Azure AI Search service with agentic retrieval available in your region. Basic tier or higher if you want managed identity support.
A Microsoft Foundry project and resource, with an LLM deployment (e.g. gpt-5-mini) and an embedding model (e.g. text-embedding-3-large).
The Search Index Data Reader role assigned to the identity that will query the knowledge base.
If your knowledge base specifies an LLM, the search service needs a managed identity with Cognitive Services User on the Foundry resource.
Python 3.10+.

Install the client libraries:

# Preview SDK — required for answer synthesis, configurable reasoning effort,
# document-level permissions and multi-turn retrieve.
pip install --pre azure-search-documents

# Stable SDK is enough if you only need GA features on 2026-04-01:
# pip install azure-search-documents

pip install azure-identity langchain-mcp-adapters langgraph langchain-openai httpx

A word on API versions before you start

This is the decision that will bite you later if you get it wrong, so make it deliberately now.

Agentic retrieval is generally available in the 2026-04-01 REST API. The 2026-05-01-preview adds answer synthesis, configurable reasoning effort, the messages input, document-level permissions and sensitivity-label metadata. Both the Azure portal and the Microsoft Foundry portal expose preview-only behaviour regardless of what your code uses, which means the portal is not a reliable preview of what your production code will do.

The API version also changes MCP behaviour directly. With 2026-05-01-preview, the knowledge base can return synthesized answers when it is configured with an LLM and a compatible reasoning effort. With 2026-04-01, MCP retrieval is always minimal and extractive, and the connection returns grounding data only.

Pick 2026-04-01 if you are shipping to production now and can live with extractive grounding. Pick 2026-05-01-preview if you need synthesis or permission filtering, and accept there is no SLA.

Step 1 — Create the knowledge source and knowledge base

A knowledge source points at your content. A knowledge base wraps one or more sources with retrieval configuration. Create the source first.

import os
from azure.identity import DefaultAzureCredential
from azure.search.documents.indexes import SearchIndexClient
from azure.search.documents.indexes.models import (
    KnowledgeBase,
    KnowledgeSourceReference,
    KnowledgeBaseAzureOpenAIModel,
    AzureOpenAIVectorizerParameters,
    SearchIndexKnowledgeSource,
    SearchIndexKnowledgeSourceParameters,
)

SEARCH_ENDPOINT = os.environ["AZURE_SEARCH_ENDPOINT"]   # https://<svc>.search.windows.net
AOAI_ENDPOINT   = os.environ["AZURE_OPENAI_ENDPOINT"]
CHAT_DEPLOYMENT = "gpt-5-mini"

credential = DefaultAzureCredential()
index_client = SearchIndexClient(endpoint=SEARCH_ENDPOINT, credential=credential)

# A knowledge source over an index you already have.
knowledge_source = SearchIndexKnowledgeSource(
    name="product-docs-ks",
    description=(
        "Product documentation, release notes and API reference. "
        "Use for questions about product behaviour, configuration and limits."
    ),
    search_index_parameters=SearchIndexKnowledgeSourceParameters(
        search_index_name="product-docs-index",
        source_data_select="id,title,content,url,updated_at",
    ),
)
index_client.create_or_update_knowledge_source(knowledge_source=knowledge_source)

The description is not decoration. The retrieval engine uses it when deciding which sources to query for a given subquery, so write it the way you would explain the source to a new colleague: what is in it, and what kinds of question it answers.

Now the knowledge base:

knowledge_base = KnowledgeBase(
    name="product-docs-kb",
    description="Grounding for product support questions.",
    knowledge_sources=[
        KnowledgeSourceReference(name="product-docs-ks"),
    ],
    models=[
        KnowledgeBaseAzureOpenAIModel(
            azure_open_ai_parameters=AzureOpenAIVectorizerParameters(
                resource_url=AOAI_ENDPOINT,
                deployment_name=CHAT_DEPLOYMENT,
                model_name=CHAT_DEPLOYMENT,
            )
        )
    ],
    retrieval_instructions=(
        "Prefer the most recently updated documents when versions conflict. "
        "For questions about limits or quotas, always consult product-docs-ks."
    ),
)
index_client.create_or_update_knowledge_base(knowledge_base=knowledge_base)
print("knowledge base ready")

retrieval_instructions steers the planner's source selection. It is the highest-leverage knob in the whole configuration and the one most people leave empty.

Index requirements. If you point at an existing index, it needs a semantic configuration — agentic retrieval uses L2 semantic ranking. If the index has vector fields, it also needs a valid vectorizer so the engine can vectorize subqueries; otherwise vector fields are silently ignored.

Step 2 — Verify with the retrieve API before you touch MCP

Do not debug two systems at once. Confirm retrieval works over the SDK first, where you get the full response envelope including the query plan.

from azure.search.documents.knowledgebases import KnowledgeBaseRetrievalClient
from azure.search.documents.knowledgebases.models import (
    KnowledgeBaseMessage,
    KnowledgeBaseMessageTextContent,
    KnowledgeBaseRetrievalRequest,
)

kb_client = KnowledgeBaseRetrievalClient(
    endpoint=SEARCH_ENDPOINT,
    knowledge_base_name="product-docs-kb",
    credential=credential,
)

request = KnowledgeBaseRetrievalRequest(
    messages=[
        KnowledgeBaseMessage(
            role="user",
            content=[KnowledgeBaseMessageTextContent(
                text="What are the rate limits on the ingestion API, and did they change in the last release?"
            )],
        )
    ],
    include_activity=True,   # gives you the query plan
)

result = kb_client.retrieve(request)

print(result.response[0].content[0].text[:800])   # grounding data
for entry in result.activity:                     # what the planner actually did
    print(entry.type, getattr(entry, "elapsed_ms", None))

The activity array is your observability surface. It reports the planner's token usage, the subqueries that were issued to each source, elapsed time per source, and reasoning-token consumption. Read it now, because — as we will see in Step 5 — you do not get it back over MCP.

If the response is empty but activity shows matches were found, a document probably exceeded the output budget. Increase max_output_size, or chunk large source documents more aggressively.

Step 3 — Understand what the MCP endpoint gives you (and what it does not)

Every knowledge base is automatically an MCP server. There is nothing to deploy. The endpoint is:

https://<your-search-service>.search.windows.net/knowledgebases/<your-knowledge-base>/mcp?api-version=<api-version>

It exposes exactly one tool, knowledge_base_retrieve. Clients cannot see index management or source configuration through it — the surface is deliberately narrow.

The trap is that the MCP tool result is not the retrieve action's response. It is a plain MCP tool result, which most clients surface under result.content[]:

{
  "result": {
    "content": [
      {
        "type": "text",
        "text": "[{\"ref_id\":\"0\",\"title\":\"Ingestion limits\",\"terms\":\"rate limit, throttling\",\"content\":\"<chunk>\"}]"
      }
    ]
  }
}

That text field is a JSON-encoded string, not a JSON object. You have to parse it twice. And the activity and references arrays you relied on in Step 2 are simply absent.

Here is the comparison in full — this table is the thing to keep open while you build:

Aspect	Retrieve action (REST / SDK)	MCP endpoint (`knowledge_base_retrieve`)
Payload location	`response[0].content[0].text`	`result.content[0].text`
Grounding data format	JSON-encoded string	JSON-encoded string (same inner shape)
`activity` array (query plan, tokens, per-source timings)	Returned when `includeActivity` is set	Not returned
`references` array (`ref_id` → `docKey`, `activitySource`)	Returned, controllable per source	Not returned
Sensitivity label metadata	Per-reference + response-level aggregate	Same fields surfaced when configured
Answer synthesis	Available on `2026-05-01-preview`	Available on `2026-05-01-preview` only; `2026-04-01` is always extractive
Per-request tuning (`filterAddOn`, `maxOutputDocuments`, `failOnError`)	Full control per knowledge source	Not exposed — set defaults on the knowledge base instead
Auth mechanism	SDK credential object	`Authorization: Bearer` header, or `api-key` header
Best for	Deterministic pipelines, evaluation harnesses, observability	Agent frameworks, tool-calling loops, cross-runtime reuse

The practical consequence: use MCP for the agent loop, and keep a direct retrieve client around for evaluation and debugging. They point at the same knowledge base, so there is no duplication of configuration — only of client code, and only where it earns its keep.

Step 4 — Authenticate, properly

Two options. Only one belongs in production.

Admin key (api-key header) grants full read-write access to the search service. Use it for a five-minute spike, never beyond that.

Bearer token (Authorization header) is the recommended path. The identity behind the token needs Search Index Data Reader on the search service, and the token must be scoped to https://search.azure.com/.default.

The naive version looks like this, and it works — for about an hour:

from azure.identity import DefaultAzureCredential, get_bearer_token_provider

credential = DefaultAzureCredential()
search_token_provider = get_bearer_token_provider(
    credential, "https://search.azure.com/.default"
)

MCP_URL = (
    f"{SEARCH_ENDPOINT}/knowledgebases/product-docs-kb/mcp"
    "?api-version=2026-05-01-preview"
)

connection = {
    "foundry_iq": {
        "transport": "http",              # streamable HTTP
        "url": MCP_URL,
        "headers": {"Authorization": f"Bearer {search_token_provider()}"},
    }
}

Note the parentheses: search_token_provider() is evaluated once, at construction time, and frozen into a dict. A long-running graph, a checkpointed conversation resumed the next morning, or a service that builds its client at startup will all start returning 401s once that token expires.

The fix: an `httpx.Auth` that refreshes

langchain-mcp-adapters uses the official MCP SDK underneath, which accepts a custom authentication mechanism implementing the httpx.Auth interface. That is where token refresh belongs.

import httpx
from azure.identity import DefaultAzureCredential, get_bearer_token_provider


class EntraBearerAuth(httpx.Auth):
    """Attaches a fresh Entra ID bearer token to every MCP request.

    azure-identity caches the token internally and only round-trips to
    the IdP when it is close to expiry, so calling the provider per
    request is cheap.
    """

    def __init__(self, credential, scope: str = "https://search.azure.com/.default"):
        self._provider = get_bearer_token_provider(credential, scope)

    def auth_flow(self, request: httpx.Request):
        request.headers["Authorization"] = f"Bearer {self._provider()}"
        yield request


auth = EntraBearerAuth(DefaultAzureCredential())

connection = {
    "foundry_iq": {
        "transport": "http",
        "url": MCP_URL,
        "auth": auth,          # instead of a frozen headers dict
    }
}

This single change is the difference between a demo and something you can leave running.

Step 5 — Load the tool into LangGraph

With authentication sorted, wiring the tool in is short.

from langchain_mcp_adapters.client import MultiServerMCPClient

client = MultiServerMCPClient(connection)
tools = await client.get_tools()

print([t.name for t in tools])   # ['knowledge_base_retrieve']

For a quick check, hand the tool straight to a prebuilt agent:

from langgraph.prebuilt import create_react_agent
# (LangChain v1 equivalent: from langchain.agents import create_agent)

agent = create_react_agent("azure_openai:gpt-5-mini", tools)

response = await agent.ainvoke(
    {"messages": [{"role": "user", "content":
        "What changed about ingestion rate limits in the last release?"}]}
)
print(response["messages"][-1].content)

If that returns a grounded answer, the integration works. Two operational notes before you build the real graph:

MultiServerMCPClient is stateless by default — each tool invocation opens a fresh MCP session and tears it down. For a stateful server you would use client.session(), but for Foundry IQ retrieval, stateless is correct and cheaper.
If you register several MCP servers, be aware that a single failing server has historically been able to take down get_tools() for all of them. Register Foundry IQ in its own client if the rest of your tool estate is flaky.

Step 6 — Build the graph, and parse the response

The prebuilt agent hides the thing you most need to control: what happens to the grounding data on its way into state. Here is the explicit version.

Figure 2 — Two planners run per turn. Step 2 decides whether to retrieve; step 5 decides how.

import json
from typing import Annotated, TypedDict

from langchain_core.messages import ToolMessage
from langchain_openai import AzureChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode


class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    citations: list          # accumulated across the conversation


SYSTEM = (
    "You answer questions about our product documentation. "
    "Always call knowledge_base_retrieve before answering a factual question. "
    "Cite sources using the ref_id values in the grounding data. "
    "If the grounding data does not contain the answer, say you do not know."
)

llm = AzureChatOpenAI(azure_deployment="gpt-5-mini", api_version="2025-04-01-preview")
llm_with_tools = llm.bind_tools(tools)


async def agent_node(state: AgentState):
    messages = [{"role": "system", "content": SYSTEM}] + state["messages"]
    return {"messages": [await llm_with_tools.ainvoke(messages)]}


def parse_grounding(state: AgentState):
    """Pull ref_id/title/url out of the last tool message into state.

    The MCP tool result is a JSON-encoded string inside a text content
    block, so it needs a second json.loads().
    """
    last = state["messages"][-1]
    if not isinstance(last, ToolMessage):
        return {}

    raw = last.content
    if isinstance(raw, list):                     # content-block form
        raw = next((b.get("text", "") for b in raw
                    if isinstance(b, dict) and b.get("type") == "text"), "")

    try:
        docs = json.loads(raw)
    except (json.JSONDecodeError, TypeError):
        return {}                                 # synthesized answer, not extractive

    found = [
        {
            "ref_id": d.get("ref_id"),
            "title": d.get("title"),
            "url": d.get("url"),
        }
        for d in docs
        if isinstance(d, dict)
    ]
    return {"citations": state.get("citations", []) + found}


def should_continue(state: AgentState):
    last = state["messages"][-1]
    return "tools" if getattr(last, "tool_calls", None) else END


builder = StateGraph(AgentState)
builder.add_node("agent", agent_node)
builder.add_node("tools", ToolNode(tools))
builder.add_node("parse", parse_grounding)

builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", should_continue, {"tools": "tools", END: END})
builder.add_edge("tools", "parse")
builder.add_edge("parse", "agent")

graph = builder.compile()

The parse node is what keeps citations alive. Without it, the grounding JSON passes through the message history as an opaque blob, the model cites ref_id values, and your UI has nothing to resolve them against.

Note the defensive try/except: if you switch the knowledge base to answer-synthesis output mode, the tool returns prose rather than a JSON array, and a parser that assumes JSON will crash on a configuration change made by someone else in the Azure portal.

Step 7 — Enforce per-user permissions

Everything so far runs as a single service identity, which means every user sees every document the service can see. For most enterprise deployments that is unacceptable.

Permission enforcement has two halves.

At ingestion time, indexed sources need ingestionPermissionOptions set so that ACLs, RBAC scopes or Purview sensitivity labels are ingested alongside content. If you skip this, results come back unfiltered no matter what you send at query time — and the only fix is to recreate the knowledge source. Federated sources work differently: remote SharePoint queries through the Copilot Retrieval API using the user's own token and never ingests anything, and Fabric and Work IQ sources exchange the user's token for a scoped one.

At query time, you pass the end user's access token — scoped to https://search.azure.com/.default, separate from your service credential, and requiring no search-service permissions of its own — in the x-ms-query-source-authorization header.

Over the SDK that is a named parameter:

result = kb_client.retrieve(
    retrieval_request=request,
    x_ms_query_source_authorization=user_token,   # the end user, not the service
)

Over MCP, it is a per-request header, which means it varies per user while your client is long-lived. Extend the auth class rather than rebuilding the client:

import contextvars

current_user_token = contextvars.ContextVar("current_user_token", default=None)


class EntraBearerAuthWithUser(EntraBearerAuth):
    def auth_flow(self, request: httpx.Request):
        request.headers["Authorization"] = f"Bearer {self._provider()}"
        user_token = current_user_token.get()
        if user_token:
            request.headers["x-ms-query-source-authorization"] = user_token
        yield request

Set the context variable at the edge of your application — in the FastAPI dependency or middleware that already validates the caller — and every downstream MCP call in that request inherits it, including calls made deep inside a graph.

Verify this end to end in your own environment. Microsoft documents the header explicitly for the retrieve action, and notes that MCP clients configure custom headers differently. Test with two users who have genuinely different document access and confirm the result sets differ — do not assume it works because it did not error.

Step 8 — Decide who plans

This is the design question the tutorial format tends to bury, so it gets its own step.

Look again at Figure 2. Your LangGraph agent node runs an LLM to decide whether to retrieve and how to phrase the query. Then the knowledge base runs another LLM to decompose that query into subqueries and choose sources. Two planners, two model calls, two chances to lose the user's intent.

The failure mode is specific: the agent node paraphrases the user's question before handing it over, dropping a constraint ("in the 2026 release", "for part XYZ2B"), and the knowledge base then plans excellent subqueries for the wrong question. Microsoft's own evaluation work identifies exactly this — constraint preservation in the handoff from orchestrator to retriever — as the thing that correlates with retrieval quality.

Three rules that follow from it:

Pass the question through, do not summarize it. Instruct the agent node to forward the user's wording, including qualifiers, rather than composing a "better" search query. The knowledge base is better at query formulation than your agent node is; that is what you are paying it for.
Tune reasoning effort, not prompts. minimal skips LLM planning entirely and runs keyword or hybrid search on the query as given — the right choice for lookups. low and medium add planning; medium adds iterative search, where the engine reviews its own results and issues follow-ups. Answer synthesis requires low or medium. Route cheap questions to a minimal knowledge base and hard ones to a medium one, and you have turned a latency/quality trade-off into a graph edge.
Prefer extractive output inside an agent. Answer synthesis produces a finished natural-language answer, which is what you want when retrieval output goes straight to a user. Inside a LangGraph agent, the agent is going to reason over the content anyway — synthesizing first costs tokens and latency, and flattens the structure your parse node wants.

Troubleshooting

Symptom	Likely cause	What to do
`403` from Azure AI Search	Identity lacks Search Index Data Reader on the search service	Assign the role; confirm you are signed in to the right tenant and subscription
`401` after roughly an hour	Bearer token frozen into a static headers dict	Switch to the `httpx.Auth` provider from Step 4
`400 Bad Request`	A `knowledgeSourceName` is not attached to the knowledge base, or its `kind` does not match; or one option requires another that is not enabled	Read the top-level error — it names the offending property
`206 Partial Content`	At least one source failed, none of them marked required	Inspect the `activity` entries carrying an `error`; process partial results or mark the critical source `failOnError`
`502 Bad Gateway`	Every selected source failed, or a source marked `failOnError` failed	Do not assume an outage — read the underlying source failure first
Empty response, but activity shows matches	Most relevant document exceeded the output budget	Raise `maxOutputSize`, or chunk large documents at ingestion
Every user sees every document	`ingestionPermissionOptions` was not set when the knowledge source was created	Recreate the knowledge source with the right options; the header alone will not fix it
Answers ignore recent documents	Scoring profiles are not applied by agentic retrieval	Use freshness-aware retrieval rather than an index scoring profile
Tool list comes back empty	One failing server in a multi-server client	Give Foundry IQ its own `MultiServerMCPClient`

Where to go next

The same knowledge base you just built is reachable from Microsoft Agent Framework, Foundry Agent Service, GitHub Copilot, Claude and Cursor without any change to its configuration. That is the real payoff of putting retrieval behind MCP rather than inside your agent: when your team standardizes on a different runtime next year, the knowledge layer does not move.

Two things worth building next: an evaluation harness that calls the retrieve API directly (so you get the activity array and can measure whether your retrieval_instructions are actually steering source selection), and a second knowledge base at minimal reasoning effort so you can route by question difficulty.

References

Query a knowledge base using the retrieve action or MCP endpoint — the authoritative reference for the MCP endpoint URL, authentication, response shapes, permission headers and troubleshooting status codes.
Agentic retrieval in Azure AI Search — overview
Model Context Protocol (MCP) — LangChain docs — transports, custom httpx.Auth, session lifecycle.
langchain-mcp-adapters on GitHub
MultiServerMCPClient API reference
Foundry IQ: build smarter agents faster with unified knowledge and serverless retrieval — Build 2026 announcement; GA scope and the MCP server.
Foundry IQ: improve recall by up to 54% with knowledge bases — the constraint-preservation evaluation behind Step 8.

Building Meeting Audio RAG on Microsoft Foundry with Fast Transcription and Foundry IQ

Jubin Soni — Tue, 11 Aug 2026 19:47:53 +0000

Every recorded meeting your organization has ever held is already a knowledge base. It just happens to be stored in the least queryable format imaginable, which is a wall of MP4 files sitting in a storage account that nobody opens twice. The good news is that the gap between that wall of files and a working question-answering agent is now much shorter than it used to be, because Microsoft Foundry ships the two halves you need in one place. Fast transcription turns the audio into diarized text in seconds rather than in real time, and Foundry IQ turns that text into a permission-aware knowledge base that any agent can query through a single endpoint.

This walkthrough builds the whole thing end to end. By the end you will have a pipeline that watches a blob container for new recordings, transcribes them with speaker labels, chunks them into speaker turns with enough metadata to make citations useful, indexes them as a Foundry IQ knowledge source, and exposes a Foundry agent that answers questions like "what did we decide about the pricing migration in Q2 and who pushed back" with real references back to the moment in the recording.

A quick naming note before we start, because the ground has moved. At Ignite 2025 Microsoft renamed Azure AI Foundry to Microsoft Foundry, and the rename was formalized in the January 2026 Product Terms. The platform is the same platform, but there are now two portal experiences and two generations of SDK. The 2.x preview of azure-ai-projects targets the new Foundry portal and API, and the 1.x GA line targets what the docs call Foundry classic. Everything in this article uses the 2.x line and the Responses-based agent surface.

What we are building, and the shape of the data flow

The pipeline has two independent halves that meet at a blob container of curated transcripts. The ingestion half is batch and event-driven. It cares about throughput and about not losing files. The retrieval half is synchronous and user-facing. It cares about latency and about grounding quality. Keeping them decoupled through storage means you can reindex, re-chunk, or swap the retrieval strategy without touching a byte of audio again.

The flow is worth reading left to right once. A recording lands in raw-recordings. Event Grid picks up the Blob Created event and drops a message on a queue, which gives you retry semantics and a dead letter path for free. A queue-triggered Function pulls the message, POSTs the audio to the Foundry Speech fast transcription endpoint, and gets back a synchronous response containing diarized phrases. A second stage groups those phrases into speaker turns, attaches timestamps and meeting metadata, and writes JSONL into curated-transcripts. Foundry IQ indexes that container on a schedule.

Why a queue between Event Grid and the Function rather than a direct trigger? Because fast transcription is synchronous and the audio files are large. A direct blob trigger gives you very little control over concurrency, and the first time somebody bulk-uploads six months of archived recordings you will saturate your Speech resource and start collecting 429s. The queue lets you cap batchSize in host.json and shape the load.

Standing up the Foundry project and the Speech resource

Create a Foundry project first. In the portal, make sure the New Foundry toggle is on, then create or select a project. The thing you need out of the portal is the project endpoint, which has the form https://<resource-name>.services.ai.azure.com/api/projects/<project-name>.

Install the preview packages.

pip install "azure-ai-projects>=2.4.0" azure-identity openai azure-storage-blob requests
az login

Entra ID is the only authentication method the projects client supports, so there is no key-based escape hatch here. Give yourself the Azure AI User role on the project resource for development work. For the pipeline itself, use a user-assigned managed identity and grant it Azure AI User plus Storage Blob Data Contributor.

Two environment variables carry the rest of the article.

export FOUNDRY_PROJECT_ENDPOINT="https://your-account.services.ai.azure.com/api/projects/meetings"
export SPEECH_RESOURCE_NAME="your-speech-resource"

Confirm the project client talks to the service before you build anything on top of it.

import os
from azure.ai.projects import AIProjectClient
from azure.identity import DefaultAzureCredential

with (
    DefaultAzureCredential() as credential,
    AIProjectClient(
        endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
        credential=credential,
    ) as project,
):
    openai = project.get_openai_client()
    r = openai.responses.create(
        model="gpt-5-mini",
        input="Reply with the single word ready.",
    )
    print(r.output_text)

get_openai_client() returns an authenticated client from the openai package configured to run Responses operations against your Foundry project endpoint. That is the pattern to internalize. You use the project client for setup, configuration, agents, and evaluations, and the OpenAI-compatible client for the actual model calls.

Turning an hour of audio into diarized speaker turns

Fast transcription is the right tool for recorded meetings. It returns results synchronously and much faster than real time, which is exactly the tradeoff you want for a file that already exists. Batch transcription is the alternative and it wins on very long archives and on advanced customization, but for a one-hour standard-format recording, fast transcription gets you a result in a small number of seconds with predictable latency.

The endpoint is /speechtotext/transcriptions:transcribe and the current generally available API version is 2025-10-15. It takes multipart/form-data with the audio in one part and a JSON definition in another. Diarization is configured with a diarization object carrying maxSpeakers, and the service can separate up to 35 distinct speakers in a single channel before it errors out.

Here is the worker in full, with the retry behavior that you will absolutely need.

import json
import os
import time
import requests
from azure.identity import DefaultAzureCredential

SPEECH_ENDPOINT = (
    f"https://{os.environ['SPEECH_RESOURCE_NAME']}"
    ".cognitiveservices.azure.com/speechtotext/transcriptions:transcribe"
    "?api-version=2025-10-15"
)
SCOPE = "https://cognitiveservices.azure.com/.default"
RETRYABLE = {408, 429, 500, 502, 503, 504}


def transcribe(audio_path, locales=("en-US",), max_speakers=8, max_attempts=5):
    """Fast transcription with diarization and bounded exponential backoff."""
    credential = DefaultAzureCredential()
    definition = {
        "locales": list(locales),
        "diarization": {"enabled": True, "maxSpeakers": max_speakers},
        "profanityFilterMode": "None",
    }

    for attempt in range(max_attempts):
        token = credential.get_token(SCOPE).token
        with open(audio_path, "rb") as fh:
            response = requests.post(
                SPEECH_ENDPOINT,
                headers={"Authorization": f"Bearer {token}"},
                files={"audio": (os.path.basename(audio_path), fh)},
                data={"definition": json.dumps(definition)},
                timeout=600,
            )

        if response.status_code == 200:
            return response.json()

        if response.status_code not in RETRYABLE:
            raise RuntimeError(
                f"Fast transcription failed {response.status_code} {response.text[:400]}"
            )

        wait = float(response.headers.get("Retry-After", 2 ** attempt))
        time.sleep(min(wait, 60))

    raise RuntimeError(f"Giving up on {audio_path} after {max_attempts} attempts")

A few things in there earn their place. The Retry-After header is honored when the service sends one, which matters a lot under throttling because blind exponential backoff on a shared Speech resource just means every worker retries in lockstep. Profanity filtering is set to None because the default is Masked and masked words in a transcript quietly damage retrieval, since the asterisks become tokens that match nothing. The 600 second timeout is generous on purpose, because a large file uploading over a constrained egress path can spend a long while before the service even starts work.

The response contains a phrases array where each entry carries speaker, offsetMilliseconds, durationMilliseconds, and text. Phrases are the wrong chunk size for retrieval. They are usually a sentence or two, which means an embedding of a phrase carries almost no context and a citation to a phrase drops the reader into the middle of a thought. Group them into speaker turns instead.

from dataclasses import dataclass, asdict


@dataclass
class Turn:
    meeting_id: str
    meeting_title: str
    meeting_date: str
    speaker: str
    start_ms: int
    end_ms: int
    text: str

    @property
    def chunk_id(self):
        return f"{self.meeting_id}-{self.start_ms:09d}"


def to_turns(result, meta, max_chars=2400, gap_ms=4000):
    """Collapse diarized phrases into speaker turns, splitting very long ones."""
    turns, current = [], None

    for p in result.get("phrases", []):
        speaker = f"Speaker {p.get('speaker', 'unknown')}"
        start = p["offsetMilliseconds"]
        end = start + p["durationMilliseconds"]

        same_speaker = current and current.speaker == speaker
        contiguous = current and (start - current.end_ms) < gap_ms
        room = current and (len(current.text) + len(p["text"])) < max_chars

        if same_speaker and contiguous and room:
            current.text += " " + p["text"]
            current.end_ms = end
            continue

        if current:
            turns.append(current)
        current = Turn(
            meeting_id=meta["meeting_id"],
            meeting_title=meta["title"],
            meeting_date=meta["date"],
            speaker=speaker,
            start_ms=start,
            end_ms=end,
            text=p["text"],
        )

    if current:
        turns.append(current)
    return turns

The gap_ms guard is the part people leave out. Without it, a speaker who talks at minute three and again at minute forty gets merged into one chunk if nobody else spoke in between, which is rare but produces a chunk whose timestamp range is meaningless. Four seconds of silence is a reasonable turn boundary for meeting audio.

Making chunks that are worth citing

Retrieval quality on meeting transcripts lives or dies on what surrounds the raw text. A bare speaker turn like "yeah I think that's fine, let's go with option two" is nearly unretrievable, because it contains no nouns. The fix is to write a small amount of generated context into each record and let the hybrid search match on that.

def contextualize(openai, turn, neighbors):
    """Prepend a one-line situating summary so short turns stay retrievable."""
    window = "\n".join(f"{n.speaker}: {n.text}" for n in neighbors)
    r = openai.responses.create(
        model="gpt-4.1-mini",
        input=(
            "Write one sentence, under 25 words, situating the final utterance "
            "inside this meeting excerpt. Name the topic and any decision. "
            "Do not editorialize.\n\n"
            f"Meeting: {turn.meeting_title} ({turn.meeting_date})\n\n"
            f"{window}\n\nFinal utterance: {turn.speaker}: {turn.text}"
        ),
    )
    return r.output_text.strip()


def to_records(openai, turns):
    for i, turn in enumerate(turns):
        neighbors = turns[max(0, i - 3): i + 1]
        context = contextualize(openai, turn, neighbors)
        yield {
            **asdict(turn),
            "chunk_id": turn.chunk_id,
            "context": context,
            "content": f"{context}\n\n{turn.speaker}: {turn.text}",
            "timecode": f"{turn.start_ms // 60000:02d}:{(turn.start_ms // 1000) % 60:02d}",
        }

This costs one small model call per turn, which on a one-hour meeting is a few hundred calls of a couple hundred tokens each. Run it concurrently with a semaphore rather than serially. The timecode field is what makes citations feel like a product feature rather than a footnote, because you can render it as a deep link into your video player.

Write the records as JSONL to curated-transcripts, one file per meeting, and you are done with audio forever.

Wiring the transcripts into a Foundry IQ knowledge base

Foundry IQ is the knowledge and retrieval layer built on Azure AI Search. The mental model is two nested objects. A knowledge source points at searchable content, and a knowledge base wraps one or more knowledge sources behind a single endpoint that agents query. For indexed sources, Foundry IQ manages the whole indexing pipeline, so content gets ingested, chunked, vectorized, and prepared for hybrid retrieval without you standing up a skillset by hand.

Agentic retrieval features are generally available in the 2026-04-01 REST API. The 2026-05-01-preview version exposes the fuller feature set, including preview knowledge source kinds and the ability to attach an LLM to non-web sources. Blob Storage is a generally available indexed source kind, which is exactly what we need.

Point a knowledge source at the curated container.

from azure.search.documents.indexes import SearchIndexClient
from azure.search.documents.indexes.models import (
    KnowledgeBase,
    KnowledgeSourceReference,
    AzureBlobKnowledgeSource,
    AzureBlobKnowledgeSourceParameters,
)
from azure.identity import DefaultAzureCredential

index_client = SearchIndexClient(
    endpoint=os.environ["SEARCH_ENDPOINT"],
    credential=DefaultAzureCredential(),
)

source = AzureBlobKnowledgeSource(
    name="meeting-transcripts",
    description=(
        "Diarized speaker turns from recorded internal meetings, 2024 onward. "
        "Each chunk carries meeting title, date, speaker label, and timecode."
    ),
    azure_blob_parameters=AzureBlobKnowledgeSourceParameters(
        connection_string=os.environ["BLOB_CONNECTION"],
        container_name="curated-transcripts",
        embedding_model=..., # your deployed text embedding model
        chat_completion_model=..., # optional, enables verbalization
    ),
)
index_client.create_or_update_knowledge_source(source)

That description field is not decoration. When a knowledge base holds several sources, the retrieval engine plans which sources to query, and the description is the primary signal it uses to route. Write it like you are briefing a colleague who has never seen your data.

Now the knowledge base.

kb = KnowledgeBase(
    name="meetings-kb",
    knowledge_sources=[
        KnowledgeSourceReference(name="meeting-transcripts", always_query_source=False),
    ],
    retrieval_instructions=(
        "Meeting transcripts. When the user asks who said or decided something, "
        "return the speaker turns that contain the statement plus the surrounding turns. "
        "Prefer recent meetings when the question is about current state."
    ),
)
index_client.create_or_update_knowledge_base(kb)

The retrieval engine plans which sources to query and performs iterative search when the first pass does not clear its relevance bar. Iterative search depends on setting a medium retrieval reasoning effort, either on the knowledge base or per request. That single knob is also the biggest lever on both latency and spend, so treat it as a tuning parameter rather than a set-and-forget value.

Reasoning effort	What the engine does	Good fit for
Minimal	Single pass, extractive results, no query planning	Lookup-style questions where the user names the meeting
Low	Light query decomposition across sources	Most interactive chat traffic
Medium	Iterative search plus richer planning over sources	Analytical questions spanning many meetings

Giving the agent a knowledge base and a personality

With the knowledge base in place, the agent is short. Agent operations in the 2.x SDK are built on the Responses protocol, and agents are versioned objects created with create_version.

from azure.ai.projects.models import PromptAgentDefinition

INSTRUCTIONS = """You answer questions about internal meetings using only the
meeting transcript knowledge base.

Rules you follow without exception.
1. Every factual claim carries a citation naming the meeting title, date, and timecode.
2. When you cannot find support in the transcripts, say so plainly and stop.
3. Attribute statements to the speaker label exactly as it appears. Never guess a real name.
4. When speakers disagreed, surface the disagreement rather than flattening it into consensus.
5. Distinguish a decision from a suggestion. Quote the language that makes it one or the other.
"""

agent = project.agents.create_version(
    agent_name="meeting-analyst",
    definition=PromptAgentDefinition(
        model="gpt-5-mini",
        instructions=INSTRUCTIONS,
        tools=[{"type": "knowledge_base", "knowledge_base": {"name": "meetings-kb"}}],
    ),
)
print(agent.id, agent.version)

Rule three is doing real work. Diarization gives you stable speaker identifiers within a recording, not identities, so you get generic labels rather than names. If the instructions do not forbid it, a capable model will cheerfully infer that Speaker 2 is the person whose name appears in the meeting title, and it will be wrong roughly as often as it is right. If you need real names, map them yourself in the chunking stage from calendar metadata or from multichannel capture, and write the resolved name into the record.

Calling the agent looks like any Responses call.

def ask(openai, agent_name, question, previous_response_id=None):
    return openai.responses.create(
        extra_body={"agent": {"name": agent_name, "type": "agent_reference"}},
        input=question,
        previous_response_id=previous_response_id,
    )


first = ask(openai, "meeting-analyst",
            "What did we decide about the pricing migration, and did anyone object?")
print(first.output_text)

follow_up = ask(openai, "meeting-analyst",
                "Which of those objections were ever resolved?",
                previous_response_id=first.id)
print(follow_up.output_text)

Threading through previous_response_id keeps the conversation server-side, which means you are not shipping a growing transcript of the chat on every turn and you are not writing your own history store.

Failing well when retrieval or the model does not cooperate

Two failure classes matter in production and they want different handling. Transient service errors want retries. Empty or weak retrieval wants a different answer, not a retry, because running the same query again against the same index returns the same nothing.

import random
from openai import APIStatusError, APITimeoutError

TRANSIENT = {408, 409, 429, 500, 502, 503, 504}


def ask_resilient(openai, agent_name, question, attempts=4, **kwargs):
    last = None
    for i in range(attempts):
        try:
            return ask(openai, agent_name, question, **kwargs)
        except APITimeoutError as exc:
            last = exc
        except APIStatusError as exc:
            if exc.status_code not in TRANSIENT:
                raise
            retry_after = exc.response.headers.get("retry-after")
            last = exc
            if retry_after:
                time.sleep(min(float(retry_after), 30))
                continue
        time.sleep(min(2 ** i + random.random(), 30))
    raise last

Full jitter on the backoff is not optional at any real concurrency. Without it your retries synchronize into a thundering herd and you turn a brief throttle into a sustained one.

For the retrieval side, the answer is to make the agent's failure visible rather than silent. Instruction two above tells the model to say it found nothing, and you should assert on that in your evaluation set. A grounded system that admits ignorance is far more valuable than one that produces confident prose from three irrelevant chunks, and the second failure mode is much harder to notice in production because the output looks fine.

Measuring whether the thing actually works

Two separate quality questions live in this pipeline and they need separate measurement. The transcription layer has an accuracy problem measured in word error rate. The retrieval and generation layer has a groundedness problem measured by a judge model. A regression in either one looks identical from the outside, which is a good argument for measuring them apart.

Build a golden set first. A hundred or so questions written against meetings you have actually listened to is worth more than a thousand synthetic ones, because the value is in the expected answers and only a human who sat through the meeting can write those. Cover the awkward shapes deliberately. Include questions whose answer is genuinely absent so you can measure refusal behavior. Include questions that span two meetings. Include questions where two people disagreed.

{"question": "Who owned the migration rollback plan after the March review?",
 "expected": "Speaker 3 accepted ownership at 41:12 in Platform Review 2026-03-04.",
 "must_cite": "Platform Review 2026-03-04",
 "kind": "attribution"}
{"question": "What was the agreed SLA for the batch job?",
 "expected": "Not discussed in any recorded meeting.",
 "must_cite": null,
 "kind": "refusal"}

The evaluation operations live on the project client in the 2.x SDK, under properties such as evaluators, evaluation_rules, and schedules. For groundedness and relevance you use built-in judge evaluators. For word error rate you register a custom evaluator, because that one is arithmetic rather than judgment.

import jiwer

def transcript_wer(reference_text, hypothesis_text):
    transform = jiwer.Compose([
        jiwer.ToLowerCase(),
        jiwer.RemovePunctuation(),
        jiwer.RemoveMultipleSpaces(),
        jiwer.Strip(),
        jiwer.ReduceToListOfListOfWords(),
    ])
    return jiwer.wer(reference_text, hypothesis_text,
                     truth_transform=transform, hypothesis_transform=transform)

Hand-correct twenty minutes of audio across three or four recordings and keep it as your reference. Twenty minutes sounds thin and it is, but it catches the failures that matter, which are domain vocabulary and acronyms coming back as phonetic mush. If your WER on product names is bad, the fix is a phrase list rather than a better model. Phrase lists let you hand the recognizer a set of words likely to appear, and they move the needle hard on proper nouns and internal jargon.

The metrics worth gating a deploy on are these four.

Metric	What it catches	Where it comes from
Word error rate on domain terms	Vocabulary drift, new product names, bad audio	Custom evaluator against hand-corrected reference
Groundedness	Answers not supported by retrieved chunks	Built-in judge evaluator
Citation validity	Fabricated meeting titles, timecodes outside the recording	Deterministic check against chunk metadata
Refusal rate on absent answers	Confident invention when nothing was retrieved	Golden set questions with no supporting content

Citation validity is the cheap one everyone skips. You already have the chunk metadata, so parsing the citations out of the answer and asserting that each meeting title exists and each timecode falls inside that recording's duration is maybe thirty lines of code. It catches a specific and embarrassing failure that judge models are surprisingly forgiving of.

Getting this to production without regrets

Reindex on a schedule and expect churn. Foundry IQ triggers indexing and data synchronization automatically for indexed sources, but your curated container is the contract. If you change chunking strategy, you are rewriting every record, and a full reindex of a large corpus is not instant. Version your chunking logic and write the version into each record so you can tell mixed-generation content apart during a migration.

Decide the permission model before you index anything. Meeting recordings are among the most sensitive content an organization has. Retrieval in Foundry IQ respects user permissions for supported knowledge source types, and for the remote SharePoint source, Purview sensitivity labels and data classifications flow through the indexing and retrieval pipeline. Blob-backed sources do not give you that for free. If access control per meeting matters, either enforce it with security filters at query time using a field on each chunk, or keep recordings in SharePoint and use the remote source, where content never leaves SharePoint and SharePoint enforces permissions. Retrofitting this later means reindexing everything and auditing every conversation that already happened.

Instrument with tracing from day one. The projects SDK ships GenAI tracing instrumentation, currently an experimental preview where spans and attributes may change between versions. Turn it on anyway. When a user says the agent gave a bad answer, you want the retrieved chunk IDs and the query plan from that exact response, and reconstructing them after the fact from logs you did not write is miserable.

Watch the two meters. Retrieval bills token usage for subquery execution and semantic reranking, and the model you attach for query planning and answer synthesis bills separately on the model side. Reasoning effort, source count, and how much content you route into synthesis are the levers, in that order.

Plan the migration if you are on the old pattern. If you are still using Azure OpenAI On Your Data, the "Add your data" flow in the classic chat playground, it is deprecated and retires on October 14, 2026. The official migration target is exactly the stack in this article, which is Foundry Agent Service plus Foundry IQ.

How this compares to rolling the pipeline yourself

The obvious alternative is a hand-built stack. Whisper for transcription behind your own GPU or an inference endpoint, pyannote for diarization, your own chunker, a vector database, and LangChain or a custom orchestrator on top. That stack is genuinely good and it is genuinely more work. The honest comparison looks like this.

Concern	Foundry with fast transcription and Foundry IQ	Self-hosted Whisper plus pyannote plus a vector DB	Amazon Transcribe plus Bedrock Knowledge Bases	Google Speech-to-Text plus Vertex AI Search
Diarization	Built into the same call, up to 35 speakers	Separate model, separate tuning, best-in-class quality achievable	Built into the transcription job	Built into the recognizer
Time to first working answer	Hours	Days to weeks	Hours	Hours
Retrieval planning	Agentic, multi-query, iterative at higher effort	Whatever you write	Managed retrieval, less query planning	Managed retrieval with good semantic ranking
Permission-aware retrieval	Native for supported sources, Purview labels honored for remote SharePoint	You build it	IAM-scoped, coarser at the chunk level	IAM-scoped
Where the audio goes	Your Azure region	Wherever you run it, including fully on-premises	Your AWS region	Your GCP region
Escape hatch	Knowledge bases callable from any app through the Search APIs	Total control	Bedrock APIs	Vertex APIs

The self-hosted path wins on two things and they are not small. One is cost at very high volume, because at some point per-minute transcription pricing loses to a GPU you already own. The other is data residency in the strict sense, meaning audio that legally cannot leave your premises. If neither applies to you, the managed path buys back weeks of work you would otherwise spend on chunking heuristics and retry logic.

Within Azure there is also a smaller decision, which is fast transcription against batch transcription. Fast wins on latency and simplicity for files under the size limit. Batch wins when you need to process very large archives asynchronously, when you want webhook notifications on completion, or when you want to bring your own storage account for the outputs.

Where to take it next

The pipeline above is the spine. The interesting extensions hang off the chunking stage, because that is where you decide what the retrieval layer is even capable of answering. Extracting action items into a structured field lets you answer "what did I commit to last month" without any retrieval creativity. Writing a sentiment or disagreement flag onto each turn lets the agent find contested moments directly. Adding a second knowledge source pointed at your specs and design docs turns "what did we decide" into "what did we decide and does the shipped code match", and because a knowledge base fronts multiple sources behind one endpoint, that is a configuration change rather than an architecture change.

The part worth protecting as you extend is the evaluation loop. Meeting corpora grow continuously and unevenly, and a retrieval strategy tuned on six months of transcripts behaves differently on three years. The golden set is what tells you when that has happened.

References

Databricks Lakebase: Give Your Agent a Branch, Not Your Production Database

Jubin Soni — Sun, 02 Aug 2026 19:49:56 +0000

There's an awkward moment in every agentic data project. The agent works. It writes decent SQL, it reasons about schema, it proposes a migration that looks right. And then somebody asks the question nobody wants to answer: what happens when it's wrong against production?

The usual answers are all bad.

Point it at production with read-only credentials, and you've capped the damage but also capped the agent — it can diagnose and never fix. Point it at a seeded staging database, and you get an agent that's confidently correct about data that stopped resembling production four months ago. Spin up a container per run, and you're maintaining fixture pipelines forever while paying for compute that sits idle between runs.

What you actually want is a full copy of production that appears in under a second, costs almost nothing while it exists, and vanishes without a trace when the agent is done. Which is to say: you want git checkout -b for your database.

Databricks Lakebase does this, and the mechanism is interesting enough to be worth understanding rather than just enabling.

How Lakebase gets there

Lakebase is Postgres — genuinely Postgres, not a wire-compatible reimplementation. What Databricks changed is the part underneath.

Standard Postgres is a monolith with compute and storage tightly coupled. Databricks decoupled the two and moved storage onto lake storage, which creates an immediate problem: object storage has high latency and no transactional consistency, and Postgres assumes neither. Two components patch the gap. Safekeepers, built on the Paxos consensus algorithm, handle low-latency durable writes. Page servers handle low-latency reads, materializing pages on demand from the lake.

Once storage is decoupled and page-addressed, branching becomes almost free. It's copy-on-write: the data stays in one place on the lake, and Lakebase tracks only the deltas between branches. Writing to a branch records the change independently without touching the others.

The numbers that matter for agent work: a branch takes roughly 500 milliseconds, a new instance comes up in under 500 milliseconds, rollback to an earlier snapshot is equally fast, and idle branches scale to zero so you pay only cheap lake storage while they sit around. Databricks reports 12 million database launches per day in production, which is less a benchmark than an indication that ephemeral instances are the intended usage pattern rather than an edge case.

That last property is what makes this viable for agents specifically. As Ali Ghodsi put it on the summit keynote stage, agents "don't want to wait 10 minutes on a database to come up." An agent that has to wait ten minutes for an environment will simply be given production instead, because the alternative makes the loop unusable. Sub-second provisioning is what makes the safe option also the convenient one.

Isolation strategies, compared

Approach	Provisioning time	Data realism	Idle cost	Blast radius of a bad write	Ongoing maintenance
Read-only prod credentials	None	Perfect	None	Zero — but agent can't act	None
Full prod write access	None	Perfect	None	Catastrophic	None
Seeded staging database	Minutes to hours	Degrades continuously	Full instance cost	Contained, but shared between runs	High — fixture pipelines, refresh jobs
Container per run with dump restore	Minutes, scales with data size	Good at restore time	None between runs	Contained per run	Medium — dump/restore tooling
Lakebase branch per run	~500 ms	Identical to parent at branch time	Near zero (storage deltas only)	Contained per run, discardable	Low

The column that usually decides it is provisioning time, because it governs whether developers actually use the isolated path or route around it under deadline pressure.

Building the loop

Here's the pattern end to end. The Databricks-specific surface is deliberately isolated into one thin layer; everything else is ordinary Postgres, because that's what it is.

Step one — get an instance. Lakebase is now included in Databricks Free Edition alongside Genie Code, serverless GPUs, Agent Bricks and LakeFlow Designer, so this costs nothing to follow along with.

# lakebase_control.py — the only Databricks-specific layer in this design.
# Endpoint shapes for branch operations are still settling; check current
# docs before shipping. Everything downstream of get_dsn() is plain Postgres.

import os
import time
import requests

WORKSPACE = os.environ["DATABRICKS_HOST"].rstrip("/")
TOKEN = os.environ["DATABRICKS_TOKEN"]
HEADERS = {"Authorization": f"Bearer {TOKEN}"}


def create_branch(parent: str, name: str) -> dict:
    """Branch an existing Lakebase instance. Copy-on-write; returns in ~500ms."""
    started = time.perf_counter()
    resp = requests.post(
        f"{WORKSPACE}/api/2.0/database/instances/{parent}/branches",
        headers=HEADERS,
        json={"name": name},
        timeout=30,
    )
    resp.raise_for_status()
    elapsed_ms = (time.perf_counter() - started) * 1000
    print(f"branch {name!r} ready in {elapsed_ms:.0f} ms")
    return resp.json()


def delete_branch(parent: str, name: str) -> None:
    requests.delete(
        f"{WORKSPACE}/api/2.0/database/instances/{parent}/branches/{name}",
        headers=HEADERS,
        timeout=30,
    ).raise_for_status()


def get_dsn(instance: str, branch: str) -> str:
    """Standard Postgres connection string for a branch."""
    resp = requests.get(
        f"{WORKSPACE}/api/2.0/database/instances/{instance}/branches/{branch}",
        headers=HEADERS,
        timeout=30,
    )
    resp.raise_for_status()
    host = resp.json()["read_write_dns"]
    return f"postgresql://{os.environ['LAKEBASE_USER']}:{TOKEN}@{host}:5432/databricks_postgres?sslmode=require"

Step two — wrap it so the agent physically cannot see production. This is the important part, and it's boring on purpose. The agent never receives a DSN; it receives a session bound to a branch.

# agent_sandbox.py
import contextlib
import psycopg
from lakebase_control import create_branch, delete_branch, get_dsn

PROD_INSTANCE = "orders-prod"


@contextlib.contextmanager
def branched_session(run_id: str, keep_on_failure: bool = True):
    """
    Yield a Postgres connection on a throwaway branch of production.

    On success the branch is discarded. On failure it is retained so you can
    attach a psql session and see exactly what the agent did — which is the
    single most useful debugging affordance in this whole design.
    """
    branch = f"agent-{run_id}"
    create_branch(PROD_INSTANCE, branch)
    conn = None
    try:
        conn = psycopg.connect(get_dsn(PROD_INSTANCE, branch), autocommit=False)
        yield conn
        conn.commit()
        delete_branch(PROD_INSTANCE, branch)
    except Exception:
        if conn:
            conn.rollback()
        if not keep_on_failure:
            delete_branch(PROD_INSTANCE, branch)
        else:
            print(f"retained branch {branch!r} for inspection")
        raise
    finally:
        if conn:
            conn.close()

Step three — verify before you believe anything. An agent reporting success is not evidence. Snapshot the invariants you care about before and after, and diff them.

# verify.py
from dataclasses import dataclass

INVARIANTS = {
    "order_count":        "SELECT count(*) FROM orders",
    "open_order_total":   "SELECT coalesce(sum(total_cents), 0) FROM orders WHERE status = 'open'",
    "orphaned_items":     "SELECT count(*) FROM order_items i "
                          "LEFT JOIN orders o ON o.id = i.order_id WHERE o.id IS NULL",
    "negative_totals":    "SELECT count(*) FROM orders WHERE total_cents < 0",
    "duplicate_skus":     "SELECT count(*) FROM ("
                          "  SELECT sku FROM products GROUP BY sku HAVING count(*) > 1"
                          ") d",
}


@dataclass
class Snapshot:
    values: dict

    @classmethod
    def take(cls, conn) -> "Snapshot":
        out = {}
        with conn.cursor() as cur:
            for name, sql in INVARIANTS.items():
                cur.execute(sql)
                out[name] = cur.fetchone()[0]
        return cls(out)

    def diff(self, other: "Snapshot") -> dict:
        return {
            k: (self.values[k], other.values[k])
            for k in self.values
            if self.values[k] != other.values[k]
        }


def assert_no_corruption(before: Snapshot, after: Snapshot) -> None:
    """Hard invariants must not move regardless of what the task was."""
    for guard in ("orphaned_items", "negative_totals", "duplicate_skus"):
        if after.values[guard] != before.values[guard]:
            raise AssertionError(
                f"invariant {guard} violated: "
                f"{before.values[guard]} -> {after.values[guard]}"
            )

Step four — the run itself.

# run_agent_task.py
import uuid
from agent_sandbox import branched_session
from verify import Snapshot, assert_no_corruption

def run(task: str, agent) -> dict:
    run_id = uuid.uuid4().hex[:8]

    with branched_session(run_id) as conn:
        before = Snapshot.take(conn)

        # The agent gets a live connection to real-shaped data and full write
        # access — to a branch that exists only for this run.
        agent.execute(task, conn)

        after = Snapshot.take(conn)
        assert_no_corruption(before, after)

        return {
            "run_id": run_id,
            "changed": after.diff(before),
        }

The full cycle

This isn't a pattern I invented for the article — it's the shape Databricks uses internally for Genie ZeroOps, its autonomous pipeline-remediation agent. ZeroOps investigates a failure, drafts a fix, then creates shallow clones of production data using the same branching mechanism as Lakebase, deploys the proposed fix to the clone, verifies row counts, and presents the result as a pull request. Nothing reaches production without explicit human approval. The verification-on-a-clone step is what makes autonomous remediation defensible rather than reckless, and it's available to you as a primitive.

What branching does not solve

Worth being direct about, because the isolation is narrower than it feels.

Side effects escape the branch. Your database is isolated. The payment API your agent calls is not. Neither is the email it sends, the Slack message it posts, or the Kafka topic it publishes to. A branch protects state you own in Postgres and nothing else — every external integration needs its own sandboxing, and this is where teams get burned after assuming the branch covered them.

Branch sprawl is real. Sub-second creation plus retain-on-failure produces a lot of branches. Enforce TTLs from day one, tag branches with the run that created them, and put a reaper on a schedule. Copy-on-write storage is cheap, not free, and a branch that diverges heavily from its parent stops being cheap.

Fresh at branch time, stale immediately after. A long-running agent task is reasoning about a snapshot. If it takes twenty minutes and the underlying question was time-sensitive, the conclusion may be wrong by the time it lands — not because the isolation failed but because you asked a real-time question of a point-in-time copy.

Invariant checks only catch what you thought to write down. The assert_no_corruption function above is a floor, not a ceiling. It catches structural damage. It won't catch an agent that correctly updated the wrong customer's record, because that's semantically valid and structurally clean. Human review of the diff remains load-bearing.

The broader shift

The reason this matters beyond Databricks is that it closes a gap that has quietly constrained every agentic data project: agents were either safe or useful, and picking both required infrastructure most teams weren't going to build.

Application code got this right decades ago. Nobody tests a refactor by editing production source — you branch, you break things freely, you merge what works. Databases never got that affordance because copying one was expensive enough that the ergonomics never arrived. Decoupling compute from storage is what finally made the copy cheap, and cheap copies are what make experimentation safe.

Give the agent a branch. Let it break things. Read the diff.

Lakebase branch APIs are evolving; the control-plane calls above are structurally accurate but verify endpoint and field names against current Databricks documentation before running in anger. Everything downstream of the DSN is standard Postgres and will behave exactly as you expect.

References

"Lakebase — Serverless Postgres on the Lakehouse." Databricks — https://www.databricks.com/product/lakebase
"Introducing Genie ZeroOps." Databricks Blog — https://www.databricks.com/blog/introducing-genie-zeroops
Marattha, P. "Databricks Data + AI Summit 2026 recap: Genie One, LTAP, Lakehouse//RT and every major launch." Flexera Blog, June 30 2026 — https://www.flexera.com/blog/perspectives/databricks-data-ai-summit-2026/
"Databricks Launches LTAP, the First Lake Transactional/Analytical Processing Architecture." Databricks Newsroom — https://www.databricks.com/company/newsroom/press-releases/databricks-launches-ltap-first-lake-transactionalanalytical
"Databricks Sandbox — Serverless Compute Documentation." Databricks Docs — https://docs.databricks.com/aws/en/compute/serverless/sandbox
"Introducing Genie One, Genie Ontology, and Genie Agents." Databricks Blog — https://www.databricks.com/blog/introducing-genie-one-genie-ontology-and-genie-agents

Microsoft Foundry Tool Search: Your Agent Pays a Tax on Every Tool It Never Calls

Jubin Soni — Sun, 02 Aug 2026 05:35:18 +0000

Most agents are billed for tools they don't use. Not once — on every single turn.

The mechanics are simple enough that it's easy to miss. When you give a model a set of tools, the full JSON schema for every tool goes into the request. Names, descriptions, parameter types, enum values, nested objects, the lot. The model reads all of it, picks one, and calls it. Next turn, the whole catalog goes over the wire again, because the API is stateless and the tool list is part of the request.

With eight tools this is invisible. With two hundred it dominates your input token bill, crowds out the context you actually care about, and — the part that hurts more — measurably degrades tool selection accuracy.

Microsoft Foundry shipped Tool Search at Build 2026 to address exactly this. It's worth understanding, and worth understanding beyond Foundry: the same failure mode shows up in any MCP-heavy agent, and the mitigation generalizes.

The shape of the problem

A moderately detailed tool schema runs 150–400 tokens once you account for parameter descriptions that are good enough for the model to use correctly. Cheap schemas produce bad tool calls, so teams write generous ones, which is the right call and also the expensive one.

Multiply that across a catalog and across turns:

tokens_per_turn  = base_prompt + conversation_history + (n_tools × avg_schema_tokens)
tokens_per_task  = tokens_per_turn × turns

A twelve-turn task against a 200-tool catalog at 250 tokens per schema spends roughly 600,000 input tokens on tool definitions alone. The conversation itself might be 20,000. You are paying, overwhelmingly, to re-read a catalog the model already decided against eleven times.

The token cost is the visible half. The invisible half is worse. As catalogs grow, they accumulate near-duplicates — get_customer, get_customer_profile, fetch_customer_record, lookup_account_by_customer — often from different teams, different MCP servers, different eras of the codebase. Selection accuracy falls not because the model got dumber but because you handed it a genuinely ambiguous menu.

What Tool Search changes

Instead of a flat list, Foundry exposes two meta-tools: tool_search and call_tool. The agent describes what it's trying to do, gets back a small ranked set of candidates, and invokes one.

The trade is a retrieval round-trip in exchange for not shipping the catalog. Above roughly thirty tools that trade is strongly favourable. Below it, it usually isn't — which is the first thing to be honest about before adopting it.

Strategies, compared

Tool Search isn't the only answer, and it isn't always the right one.

Strategy	Token cost per turn	Selection accuracy at scale	Added latency	Operational cost
Flat tool list	Linear in catalog size	Degrades sharply past ~50 tools	None	Trivial
Hand-partitioned catalogs per task type	Low within a partition	Good, if routing is correct	None	High — routing rules rot as tools change
Multi-agent split by domain	Low per sub-agent	Good within domains, poor across them	Handoff overhead	High — orchestration and shared state
Tool Search	Roughly flat regardless of catalog size	Depends on retrieval quality	One extra round trip	Low — index is maintained for you
Tool Search plus pinning	Flat, plus pinned schemas	Best available: hot path guaranteed, tail retrieved	Only on the tail	Low

Pinning is the part people skip and shouldn't. Foundry lets you pin critical tools so they bypass the search round-trip entirely, add context describing how your team actually thinks about a tool, and auto-pin frequently used ones. In practice a handful of tools account for most calls; pin those, retrieve everything else, and you get flat token cost without paying retrieval latency on the common path.

Building it end to end

Two commands scaffold a hosted agent with a toolbox attached.

mkdir my-toolbox-agent && cd my-toolbox-agent

azd ai agent init \
  -m "https://github.com/microsoft-foundry/foundry-samples/blob/main/samples/python/hosted-agents/agent-framework/responses/04-foundry-toolbox/agent.manifest.yaml" \
  --src src/toolbox-agent

Then create the toolbox from the sample's descriptor:

azd ai toolbox create my-toolbox --from-file ./src/toolbox-agent/toolbox.yaml

That prints a versioned MCP endpoint, which is what your agent binds to:

https://<account>.services.ai.azure.com/api/projects/<project>/toolboxes/my-toolbox/versions/1/mcp?api-version=v1

One gotcha worth flagging because it cost me twenty minutes: azd ai toolbox create needs a local azd project and environment to run against, even when you pass --project-endpoint explicitly. If you aren't starting from azd ai agent init, run azd init --minimal first and set the endpoint into the environment:

azd init --minimal
azd env set FOUNDRY_PROJECT_ENDPOINT https://<account>.services.ai.azure.com/api/projects/<project>

Your toolbox.yaml is where the catalog and its search behaviour are declared. The shape below is illustrative — preview schemas move, so check the current sample before copying:

# toolbox.yaml — illustrative structure, verify against the current sample
name: my-toolbox
description: Order operations, customer lookup, and fulfilment tools

toolSearch:
  enabled: true
  # Pinned tools skip retrieval entirely and are always in context.
  # Keep this list short — every pin is a permanent token cost.
  pinned:
    - get_order_status
    - search_customers
  autoPin:
    enabled: true
    minCallsPerWindow: 25

tools:
  - name: get_order_status
    source: mcp
    server: orders-mcp
    # Retrieval context: describe the tool the way your team describes it,
    # including the vocabulary users actually type.
    searchContext: >
      Look up the current fulfilment state of a single order. Use when the
      user mentions an order number, tracking number, "where is my package",
      or asks whether something shipped.

  - name: issue_refund
    source: mcp
    server: billing-mcp
    searchContext: >
      Issue a full or partial refund against a completed order. Requires an
      order ID and an amount. Do not use for cancellations of unshipped
      orders — use cancel_order instead.

That last searchContext is doing real work. Retrieval quality is the entire ballgame with Tool Search, and retrieval runs against your descriptions. The explicit negative — don't use this one, use that one — is the single highest-value thing you can write there, because near-duplicates are precisely where selection fails.

A turn, in sequence

Measuring it, because you should not take my word for it

The reason this topic is worth writing about rather than just enabling is that the win is measurable, and the size of the win depends entirely on your catalog. Wire up tracing before you change anything and get a baseline.

pip install azure-ai-projects azure-identity opentelemetry-sdk azure-core-tracing-opentelemetry

from azure.core.settings import settings
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor, ConsoleSpanExporter
from azure.ai.projects.telemetry import AIProjectInstrumentor

settings.tracing_implementation = "opentelemetry"

span_exporter = ConsoleSpanExporter()
tracer_provider = TracerProvider()
tracer_provider.add_span_processor(SimpleSpanProcessor(span_exporter))
trace.set_tracer_provider(tracer_provider)

# Emits GenAI spans for every agent and model call in this process
AIProjectInstrumentor().instrument()

Set AZURE_EXPERIMENTAL_ENABLE_GENAI_TRACING=true before running, and every turn — prompts, tool calls, model responses — surfaces as a span with token usage attached. Swap ConsoleSpanExporter for the Azure Monitor exporter and the same spans land in the Foundry portal's Observability tab.

Now run the comparison. The harness below is deliberately dumb: a fixed task set, both configurations, aggregate the spans.

import os
import json
from dataclasses import dataclass, field
from statistics import mean

@dataclass
class RunResult:
    config: str
    input_tokens: list = field(default_factory=list)
    turns: list = field(default_factory=list)
    correct_tool: list = field(default_factory=list)

TASKS = [
    # (prompt, tool the model should end up calling)
    ("Where is order 88231?",                      "get_order_status"),
    ("Refund the second item on order 88231",      "issue_refund"),
    ("Cancel order 90114, it hasn't shipped",      "cancel_order"),
    ("Which customers in Ohio ordered twice?",     "search_customers"),
    ("Send the June invoice to billing@acme.com",  "email_invoice"),
]

def score_run(spans, expected_tool):
    """Pull token usage and the actually-invoked tool out of collected spans."""
    input_tokens = sum(
        s.attributes.get("gen_ai.usage.input_tokens", 0) for s in spans
    )
    invoked = [
        s.attributes.get("gen_ai.tool.name")
        for s in spans
        if s.attributes.get("gen_ai.tool.name")
    ]
    # With Tool Search the target arrives as a call_tool argument, so unwrap it.
    resolved = [
        json.loads(s.attributes["gen_ai.tool.arguments"]).get("name", n)
        if n == "call_tool" else n
        for s, n in zip(spans, invoked)
    ]
    return input_tokens, len(invoked), expected_tool in resolved

def report(results):
    for r in results:
        print(f"\n{r.config}")
        print(f"  mean input tokens/task : {mean(r.input_tokens):>8,.0f}")
        print(f"  mean turns/task        : {mean(r.turns):>8.1f}")
        print(f"  tool selection accuracy: {mean(r.correct_tool):>8.1%}")

Run it against a flat catalog, then against the same catalog with Tool Search on, then again with your top five tools pinned. Three numbers, one table, and you'll know whether this is worth doing for your catalog rather than for a catalog in a blog post.

Where it doesn't help

Being straight about the limits, because the failure modes are real:

Small catalogs get worse, not better. Under about thirty tools, you've added a round-trip and a retrieval failure mode to save tokens you weren't spending. Don't.

Retrieval misses are silent and confusing. When a flat-list agent picks the wrong tool, the trace shows it considering the right one. When Tool Search never surfaces the right tool, the trace shows an agent that behaved reasonably given what it was handed. Debugging shifts from "why did it choose badly" to "why wasn't this in the candidate set," which is a different skill and a less obvious one.

Latency moves to the wrong place. The extra round-trip lands at the start of a turn, before any useful work. For a background agent that's free. For anything a human is watching, pin aggressively.

Your descriptions are now load-bearing infrastructure. A vague searchContext used to cost you occasional bad tool calls. Now it costs you tools that are functionally invisible. Budget review time for them the way you'd budget it for an API contract.

The general lesson

Strip Foundry out of this and the principle stands: anything you send on every turn should earn its place on every turn. Tool schemas were the first thing to hit this wall because they grow with integration count, but the same audit applies to system prompts that accumulated instructions nobody has read in six months, few-shot examples kept for a model version you no longer run, and retrieved context that's pasted in wholesale because trimming it was somebody's Q3 task.

Tool Search is a good implementation of a boring idea — retrieve instead of broadcast. The reason to adopt it isn't that it's new. It's that you can measure the difference in an afternoon, and the number is usually larger than you expect.

Preview APIs in this area are moving quickly — the azd commands and tracing setup above are current as of the June 2026 Foundry release, but verify schema-level details against the current samples before shipping.

References

Brady, N. "What's New in Microsoft Foundry | June 2026." Microsoft Foundry Blog — https://devblogs.microsoft.com/foundry/whats-new-in-microsoft-foundry-june-2026/
"Toolboxes and Routines in Microsoft Foundry." Microsoft Foundry Blog — https://devblogs.microsoft.com/foundry/toolbox-build-26/
"What's new in Microsoft Foundry | Build Edition." Microsoft Foundry Blog — https://devblogs.microsoft.com/foundry/whats-new-in-microsoft-foundry-build-2026/
"Build 2026: From Observability to ROI for AI Agents on Any Framework." Microsoft Foundry Blog — https://devblogs.microsoft.com/foundry/build-2026-from-observability-to-roi-for-ai-agents-on-any-framework/
"Microsoft Foundry Adds Runtime, Tooling, and Governance for Production Agents." InfoQ, June 2026 — https://www.infoq.com/news/2026/06/microsoft-foundry-agents/
"Microsoft Foundry docs: What's new." Microsoft Learn — https://learn.microsoft.com/en-us/azure/foundry/whats-new-foundry

Spark Performance Deep Dive on Databricks: Shuffle Tuning, Skew Handling, and Z-Ordering with Delta Lake + Unity Catalog

Jubin Soni — Mon, 27 Jul 2026 03:14:33 +0000

The problem with "just add more workers"

Most Spark performance issues on Databricks aren't solved by scaling the cluster — they're caused by shuffle and skew, and no amount of extra nodes fixes a badly partitioned join. This post builds a realistic pipeline (order events joined against a small dimension table, aggregated, and written to Delta Lake) from the ground up, and uses it to work through:

How Spark's shuffle actually behaves during a wide transformation
Diagnosing and fixing data skew with salting and adaptive query execution (AQE)
Laying out the resulting Delta table with Z-Ordering so downstream queries skip irrelevant files
Governing access to the whole pipeline with Unity Catalog

Architecture overview

Pipeline shape — a batch job reading raw events, joining against a dimension table, aggregating, and writing to a governed Delta table:

What happens inside a shuffle stage — this is the part most tutorials skip, and it's the key to understanding why skew hurts:

Step 1 — Set up governed tables in Unity Catalog

Everything downstream depends on tables being registered under Unity Catalog, which gives you centralized access control and lineage instead of per-workspace table grants.

-- setup.sql, run in a Databricks SQL or notebook cell
CREATE CATALOG IF NOT EXISTS retail_analytics;
CREATE SCHEMA IF NOT EXISTS retail_analytics.events;

CREATE TABLE IF NOT EXISTS retail_analytics.events.raw_orders (
  order_id STRING,
  customer_id STRING,
  product_id STRING,
  quantity INT,
  event_ts TIMESTAMP
)
USING DELTA
LOCATION 'abfss://data@<storage-account>.dfs.core.windows.net/raw_orders';

CREATE TABLE IF NOT EXISTS retail_analytics.events.dim_products (
  product_id STRING,
  category STRING,
  unit_cost DOUBLE
)
USING DELTA
LOCATION 'abfss://data@<storage-account>.dfs.core.windows.net/dim_products';

GRANT SELECT ON TABLE retail_analytics.events.raw_orders TO `analysts`;

Step 2 — Read and force a broadcast join for the small dimension table

dim_products is small, so let Spark broadcast it rather than shuffle both sides of the join.

# pipeline.py
from pyspark.sql import functions as F

orders = spark.table("retail_analytics.events.raw_orders")
products = spark.table("retail_analytics.events.dim_products")

joined = orders.join(F.broadcast(products), on="product_id", how="left")

Without the explicit broadcast hint, Spark's cost-based optimizer usually picks a broadcast join automatically for small tables, but being explicit avoids surprises when the dimension table grows past spark.sql.autoBroadcastJoinThreshold (default 10MB) without anyone noticing.

Step 3 — The aggregation that triggers a shuffle

groupBy on customer_id is a wide transformation — Spark must shuffle rows so all records for a given key land on the same reducer.

agg = (
    joined
    .groupBy("customer_id", "category")
    .agg(
        F.sum(F.col("quantity") * F.col("unit_cost")).alias("total_spend"),
        F.count("order_id").alias("order_count"),
    )
)

If one customer_id (say, a test account or a large B2B buyer) accounts for a disproportionate share of rows, this is where skew shows up: one reducer task runs for minutes while the rest of the stage finishes in seconds. You'll see this in the Spark UI as a single long-running task in an otherwise short stage.

Step 4 — Fixing skew with salting

Adaptive Query Execution (AQE) handles a lot of skew automatically in modern Databricks Runtime, but for known hot keys, explicit salting is still the most predictable fix.

from pyspark.sql import functions as F
import random

SALT_BUCKETS = 20

# Add a salt column to spread the hot key across multiple reducers
salted = joined.withColumn("salt", (F.rand() * SALT_BUCKETS).cast("int"))

partial_agg = (
    salted
    .groupBy("customer_id", "category", "salt")
    .agg(
        F.sum(F.col("quantity") * F.col("unit_cost")).alias("total_spend"),
        F.count("order_id").alias("order_count"),
    )
)

# Second pass: combine the salted partial aggregates into the final result
final_agg = (
    partial_agg
    .groupBy("customer_id", "category")
    .agg(
        F.sum("total_spend").alias("total_spend"),
        F.sum("order_count").alias("order_count"),
    )
)

This two-phase pattern — pre-aggregate on a salted key, then combine — is the same trick used inside combiners in older MapReduce systems. It trades a bit of extra shuffle for eliminating the single-reducer bottleneck.

Also worth setting explicitly rather than relying on the default of 200:

spark.conf.set("spark.sql.shuffle.partitions", "auto")  # let AQE size it dynamically
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")

Step 5 — Write to Delta Lake with optimized writes

(
    final_agg.write
    .format("delta")
    .mode("overwrite")
    .option("delta.autoOptimize.optimizeWrite", "true")
    .saveAsTable("retail_analytics.events.customer_spend_summary")
)

Optimized writes shuffle data before writing so you get fewer, better-sized files instead of many small ones — this costs some write-time latency in exchange for faster reads later.

Step 6 — Z-Order the table for the queries that matter

If most downstream queries filter on customer_id, co-locate related rows physically so Spark can skip files that can't match the filter.

OPTIMIZE retail_analytics.events.customer_spend_summary
ZORDER BY (customer_id);

Z-ordering aims to produce evenly balanced data files by row count rather than raw size, and its effectiveness depends on the column having reasonably high cardinality — Z-ordering by a low-cardinality column like category alone gives little benefit. On newer Delta Lake / Databricks Runtime versions, Liquid Clustering is generally the preferred choice for new tables since it adapts as query patterns change, while ZORDER remains relevant mainly for existing tables not yet migrated.

-- Preferred on new tables (Databricks Runtime supporting Liquid Clustering):
CREATE TABLE retail_analytics.events.customer_spend_summary
CLUSTER BY (customer_id);

Comparing shuffle-mitigation techniques

Technique	Fixes	Cost	When to use
Broadcast join	Shuffle on the large side of a join	Extra memory on executors	Small (<~10MB by default) dimension/lookup tables
AQE skew join handling	Automatic detection of skewed partitions	Minor planning overhead	Default-on; good general safety net
Manual salting	Known, severe hot keys	Extra shuffle for the two-phase aggregate	Hot keys that AQE doesn't fully resolve
Repartition by key	Uneven task distribution before a shuffle stage	One extra shuffle	Pre-shaping data before multiple downstream joins
Z-Ordering	Slow reads due to unnecessary file scans	Shuffle + rewrite during OPTIMIZE	Existing tables, high-cardinality filter columns
Liquid Clustering	Same as Z-Order, plus evolving query patterns	Ongoing background clustering	New tables on supporting runtime versions

Governance: tying it back to Unity Catalog

Because every table above was created under retail_analytics, access control, audit logging, and lineage are handled centrally rather than per-cluster:

-- Restrict PII-adjacent columns without duplicating the table
CREATE VIEW retail_analytics.events.customer_spend_summary_masked AS
SELECT
  customer_id,
  category,
  total_spend,
  order_count
FROM retail_analytics.events.customer_spend_summary;

GRANT SELECT ON VIEW retail_analytics.events.customer_spend_summary_masked TO `bi_readers`;

Production considerations

Diagnose before tuning. Check the Spark UI's stage view for one long-running task among many short ones — that's the signature of skew, not just "the job is slow."
Predictive optimization can run OPTIMIZE and ANALYZE automatically on Unity Catalog managed tables, which reduces the need for scheduled maintenance jobs for many workloads.
Don't Z-Order everything. It requires shuffling and rewriting the table (or partition), so reserve it for columns that are actually filtered on frequently downstream.

References

Grounded Search Agents: Microsoft Foundry + Azure AI Search + Cosmos DB + Function Calling

Jubin Soni — Mon, 27 Jul 2026 01:58:15 +0000

Why "search" isn't enough anymore

A plain RAG pipeline — embed a query, hit a vector index, stuff the top-k chunks into a prompt — works fine for a single-turn FAQ bot. It falls apart the moment you need:

Multi-turn memory ("what about last quarter?" needs to know what "last quarter" refers to)
A decision about whether to search at all (some questions are structured lookups, not document retrieval)
Citations the caller can verify

This post builds a small but complete grounded search agent: Azure OpenAI (via Microsoft Foundry) decides when to call a search tool versus a structured data tool, Azure AI Search handles retrieval and semantic ranking over unstructured documents, and Azure Cosmos DB stores conversation state and structured order/customer metadata. The agent uses OpenAI function calling to route between the two data sources and always returns citations.

We'll build this from the ground up in Python.

Architecture overview

The system has two phases: ingestion (getting documents into a searchable index) and runtime (the agent loop that answers questions).

At runtime, the agent sits between the user and two tools, using function calling to decide which one (or both) to invoke:

Step 1 — Provision and index documents in Azure AI Search

Create the index with both a vector field (for semantic similarity) and a semantic configuration (for ranking), since combining the two consistently outperforms either alone.

# create_index.py
from azure.search.documents.indexes import SearchIndexClient
from azure.search.documents.indexes.models import (
    SearchIndex, SimpleField, SearchableField, SearchField,
    SearchFieldDataType, VectorSearch, HnswAlgorithmConfiguration,
    VectorSearchProfile, SemanticConfiguration, SemanticPrioritizedFields,
    SemanticField, SemanticSearch
)
from azure.core.credentials import AzureKeyCredential

endpoint = "https://<your-search-service>.search.windows.net"
credential = AzureKeyCredential("<admin-key>")
index_client = SearchIndexClient(endpoint, credential)

fields = [
    SimpleField(name="id", type=SearchFieldDataType.String, key=True),
    SearchableField(name="content", type=SearchFieldDataType.String),
    SimpleField(name="source_url", type=SearchFieldDataType.String),
    SearchField(
        name="content_vector",
        type=SearchFieldDataType.Collection(SearchFieldDataType.Single),
        searchable=True,
        vector_search_dimensions=1536,
        vector_search_profile_name="default-profile",
    ),
]

vector_search = VectorSearch(
    profiles=[VectorSearchProfile(name="default-profile", algorithm_configuration_name="hnsw-config")],
    algorithms=[HnswAlgorithmConfiguration(name="hnsw-config")],
)

semantic_config = SemanticConfiguration(
    name="default-semantic",
    prioritized_fields=SemanticPrioritizedFields(
        content_fields=[SemanticField(field_name="content")]
    ),
)

index = SearchIndex(
    name="support-docs-index",
    fields=fields,
    vector_search=vector_search,
    semantic_search=SemanticSearch(configurations=[semantic_config]),
)

index_client.create_or_update_index(index)
print("Index created.")

Now embed and upload chunks. Chunking matters more than most tutorials admit — 512-token chunks with ~15% overlap is a reasonable default for policy/support documents.

# ingest.py
from openai import AzureOpenAI
from azure.search.documents import SearchClient
from azure.core.credentials import AzureKeyCredential
import uuid

aoai = AzureOpenAI(
    azure_endpoint="https://<your-foundry-resource>.openai.azure.com",
    api_key="<key>",
    api_version="2024-10-21",
)
search_client = SearchClient(
    endpoint=endpoint,
    index_name="support-docs-index",
    credential=AzureKeyCredential("<admin-key>"),
)

def chunk_text(text: str, max_tokens: int = 512, overlap: int = 80):
    words = text.split()
    step = max_tokens - overlap
    return [" ".join(words[i:i + max_tokens]) for i in range(0, len(words), step)]

def embed(text: str) -> list[float]:
    resp = aoai.embeddings.create(model="text-embedding-3-small", input=text)
    return resp.data[0].embedding

def ingest_document(text: str, source_url: str):
    docs = []
    for chunk in chunk_text(text):
        docs.append({
            "id": str(uuid.uuid4()),
            "content": chunk,
            "source_url": source_url,
            "content_vector": embed(chunk),
        })
    search_client.upload_documents(documents=docs)

with open("return_policy.txt") as f:
    ingest_document(f.read(), source_url="policies/return_policy.txt")

Step 2 — Structured lookups in Cosmos DB

Not every question is a document-retrieval problem. "Where's my order?" is a point lookup. Store operational data in Cosmos DB, partitioned by customer_id for predictable query performance.

# cosmos_setup.py
from azure.cosmos import CosmosClient, PartitionKey

client = CosmosClient("<cosmos-endpoint>", credential="<cosmos-key>")
db = client.create_database_if_not_exists("retail")
orders = db.create_container_if_not_exists(
    id="orders",
    partition_key=PartitionKey(path="/customer_id"),
    offer_throughput=400,
)

orders.upsert_item({
    "id": "4471",
    "customer_id": "cust_9001",
    "status": "delayed",
    "delay_reason": "carrier weather disruption",
    "eta": "2026-08-02",
})

Step 3 — The agent loop with function calling

This is the core piece: Azure OpenAI decides, per turn, whether to call search_docs, get_order, both, or neither. We define the tools as JSON schemas and let the model pick.

# agent.py
import json
from openai import AzureOpenAI
from azure.search.documents import SearchClient
from azure.cosmos import CosmosClient

aoai = AzureOpenAI(
    azure_endpoint="https://<your-foundry-resource>.openai.azure.com",
    api_key="<key>",
    api_version="2024-10-21",
)
search_client = SearchClient(endpoint, "support-docs-index", AzureKeyCredential("<admin-key>"))
cosmos = CosmosClient("<cosmos-endpoint>", credential="<cosmos-key>")
orders_container = cosmos.get_database_client("retail").get_container_client("orders")

tools = [
    {
        "type": "function",
        "function": {
            "name": "search_docs",
            "description": "Semantic search over policy and support documents. Returns passages with citations.",
            "parameters": {
                "type": "object",
                "properties": {"query": {"type": "string"}},
                "required": ["query"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "get_order",
            "description": "Look up a customer order by ID for status, ETA, and delay reason.",
            "parameters": {
                "type": "object",
                "properties": {
                    "order_id": {"type": "string"},
                    "customer_id": {"type": "string"},
                },
                "required": ["order_id", "customer_id"],
            },
        },
    },
]

def search_docs(query: str):
    results = search_client.search(
        search_text=query,
        query_type="semantic",
        semantic_configuration_name="default-semantic",
        top=3,
    )
    return [{"content": r["content"], "source": r["source_url"]} for r in results]

def get_order(order_id: str, customer_id: str):
    item = orders_container.read_item(item=order_id, partition_key=customer_id)
    return {"status": item["status"], "reason": item.get("delay_reason"), "eta": item.get("eta")}

def run_agent(user_message: str, customer_id: str, history: list):
    messages = history + [{"role": "user", "content": user_message}]
    response = aoai.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        tools=tools,
        tool_choice="auto",
    )
    msg = response.choices[0].message

    if msg.tool_calls:
        messages.append(msg)
        for call in msg.tool_calls:
            args = json.loads(call.function.arguments)
            if call.function.name == "search_docs":
                result = search_docs(**args)
            elif call.function.name == "get_order":
                args["customer_id"] = customer_id  # enforce, don't trust model-provided id
                result = get_order(**args)
            messages.append({
                "role": "tool",
                "tool_call_id": call.id,
                "content": json.dumps(result),
            })
        final = aoai.chat.completions.create(model="gpt-4o", messages=messages)
        return final.choices[0].message.content

    return msg.content

A few things worth calling out:

Never trust model-supplied identity fields. customer_id is overwritten server-side before hitting Cosmos DB, even though the model could technically supply one — this prevents cross-tenant data leaks via prompt injection.
tool_choice="auto" lets the model skip tools entirely for small talk, avoiding unnecessary search calls.
The second chat.completions.create call (after tool results are appended) is what produces the grounded, cited final answer.

Step 4 — Session state in Cosmos DB

Conversation history should not live in application memory — store it in Cosmos DB so any instance of the service can resume a session.

# session_store.py
def load_history(session_id: str) -> list:
    container = cosmos.get_database_client("retail").get_container_client("sessions")
    try:
        item = container.read_item(item=session_id, partition_key=session_id)
        return item["messages"]
    except Exception:
        return []

def save_history(session_id: str, messages: list):
    container = cosmos.get_database_client("retail").get_container_client("sessions")
    container.upsert_item({"id": session_id, "messages": messages})

Comparing retrieval strategies

Choosing between search modes is a real architectural decision, not just a config toggle:

Retrieval Mode	Best for	Weakness
Keyword (BM25)	Exact terms, SKUs, error codes	Misses paraphrased queries
Pure vector search	Conceptual / paraphrased questions	Can surface plausible-but-wrong matches
Hybrid (keyword + vector)	General-purpose document Q&A	Slightly higher latency than either alone
Semantic ranking on top of hybrid	Conversational, nuanced queries	Adds a reranking pass; not needed for simple lookups
Structured query (Cosmos DB point read)	Known-ID lookups (orders, accounts)	Not applicable to unstructured content

For most support/document agents, hybrid retrieval with semantic ranking is the right default, reserving structured lookups for anything with a stable ID.

Production considerations

Grounding failures are silent failures. If retrieval returns weak matches, the model can still generate a fluent but wrong answer — always surface citations so users (and evaluators) can catch this.
Access control matters. If your index mixes content from different permission levels, filter search results by the caller's identity before they reach the model, not after.
Rising input tokens. Retrieved passages add to the prompt on every call — budget for this in both cost and latency.

References

Building Production-Grade Semantic Search with GPT-5 and Microsoft Foundry, From Scratch

Jubin Soni — Sun, 19 Jul 2026 03:40:11 +0000

Most "RAG tutorials" stop at a single embedding query against a single index. That works for a demo and falls over the moment a real user asks something like "compare our Q3 and Q4 vendor contracts and flag anything that changed" — a question that needs multiple sub-queries, reasoning about what's still missing, and synthesis across documents.

Microsoft Foundry's answer to this is Foundry IQ: an agentic retrieval layer built on Azure AI Search that treats retrieval as a reasoning task rather than a single keyword or vector lookup. This is a from-scratch build of a semantic search pipeline using GPT-5 for query planning/synthesis and Foundry IQ for retrieval.

Architecture

Instead of one query hitting one index once, Foundry IQ's knowledge base plans sub-queries, executes them in parallel against one or more knowledge sources, evaluates whether it has enough signal, and iterates before synthesizing a final, cited answer.

The knowledge base sits between your agent and the underlying content. Your Foundry agent doesn't talk to Azure AI Search directly — it calls the knowledge base's MCP endpoint, which handles planning, retrieval, and synthesis behind a single tool call.

Step 1: Create a knowledge source

A knowledge source is a reusable reference to your underlying content — in this example, a Blob Storage container of documents. Creating it also triggers Azure AI Search to generate the index, skillset, and indexer needed to chunk and vectorize the content automatically.

from azure.search.documents.indexes import SearchIndexClient
from azure.search.documents.indexes.models import (
    SearchIndexKnowledgeSource,
    SearchIndexKnowledgeSourceParameters,
)
from azure.identity import DefaultAzureCredential

index_client = SearchIndexClient(
    endpoint="https://<your-search-service>.search.windows.net",
    credential=DefaultAzureCredential(),
)

knowledge_source = SearchIndexKnowledgeSource(
    name="vendor-contracts-ks",
    search_index_parameters=SearchIndexKnowledgeSourceParameters(
        search_index_name="vendor-contracts-index",
    ),
)
index_client.create_or_update_knowledge_source(knowledge_source=knowledge_source)

Step 2: Create a knowledge base with GPT-5 for planning and synthesis

The knowledge base ties one or more knowledge sources together with an LLM deployment that handles query planning and answer synthesis.

from azure.search.documents.indexes.models import (
    KnowledgeBase,
    KnowledgeBaseAzureOpenAIModel,
    AzureOpenAIVectorizerParameters,
)

knowledge_base = KnowledgeBase(
    name="vendor-contracts-kb",
    knowledge_sources=[{"name": "vendor-contracts-ks"}],
    models=[
        KnowledgeBaseAzureOpenAIModel(
            azure_open_ai_parameters=AzureOpenAIVectorizerParameters(
                resource_url="https://<your-foundry-resource>.openai.azure.com",
                deployment_name="gpt-5-mini",
                model_name="gpt-5-mini",
            )
        )
    ],
    output_configuration={
        "modality": "answerSynthesis",  # verbatim extractive data is the alternative
    },
)
index_client.create_or_update_knowledge_base(knowledge_base=knowledge_base)

output_configuration is the key lever here: answerSynthesis returns a pre-generated, cited answer, while extractive mode returns verbatim source chunks and leaves reasoning entirely to your agent's own model. Extractive mode costs less and gives your agent more control; synthesis mode does more work up front at the retrieval layer.

Step 3: Wire the knowledge base into a Foundry agent via MCP

Each knowledge base exposes a standalone MCP endpoint. Any MCP-compatible client — including Foundry Agent Service, but also GitHub Copilot or other MCP clients — can call its knowledge_base_retrieve tool directly.

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

project = AIProjectClient(
    endpoint="https://<your-foundry-project-endpoint>",
    credential=DefaultAzureCredential(),
)

agent = project.agents.create_agent(
    model="gpt-5-mini",
    name="contract-search-agent",
    instructions="Answer questions about vendor contracts using the knowledge base tool. Always cite sources.",
    tools=[{
        "type": "mcp",
        "server_url": "https://<your-search-service>.search.windows.net/knowledgebases/vendor-contracts-kb/mcp?api-version=2026-05-01-preview",
        "allowed_tools": ["knowledge_base_retrieve"],
    }],
)

Reasoning effort and cost/latency tradeoffs

The knowledge base's retrieval_reasoning_effort setting controls how much LLM-driven planning happens before retrieval, and it's the main dial for balancing latency, cost, and answer quality.

Reasoning effort	What happens	Best for
Minimal	Bypasses LLM query planning entirely; direct hybrid search	Simple factual lookups, latency-sensitive paths
Medium	LLM reformulates and may decompose the query into sub-queries	Multi-part or ambiguous questions
High	Full iterative planning, evaluates sufficiency, re-queries as needed	Complex, multi-hop questions across many documents

Lowering reasoning effort is also the primary way to control the number of GPT-5 tokens consumed per query — fewer planning passes and less iteration directly reduce both latency and inference cost, without needing to touch the underlying index.

Multi-hop queries in practice

The reason this matters versus classic RAG: a query like "which vendors had payment terms that changed between the Q3 and Q4 renewals" can't be answered by a single embedding lookup. With medium or high reasoning effort, the knowledge base decomposes it into sub-queries (find Q3 renewals, find Q4 renewals, compare payment terms fields), runs them in parallel against the knowledge source, and only synthesizes a final answer once it judges the retrieved context sufficient — re-querying automatically if it isn't.

References

Building a Stateful Multi-Agent System with LangGraph and Microsoft Foundry Agent Service, From Scratch

Jubin Soni — Sun, 19 Jul 2026 03:27:09 +0000

LangGraph gives you fine-grained control over agent state and routing logic in code. Microsoft Foundry gives you managed hosting, identity, tracing, and one-click publishing to Teams and M365 Copilot. Increasingly, teams want both: LangGraph's graph-based orchestration for the hard routing logic, and Foundry's operational layer for everything downstream of "it works on my machine."

This is a from-scratch walkthrough of wiring the two together — building a small multi-agent support triage system in LangGraph, then hosting it as a Foundry Hosted Agent.

Architecture

The system has three LangGraph nodes: a triage node that classifies the incoming request, and two specialist nodes (billing, technical) that the triage node routes to conditionally.

Once this graph is compiled, it doesn't run as a standalone script — it gets wrapped and exposed through Foundry's hosting layer, which handles the runtime, session management, and identity so the graph itself stays free of infrastructure concerns.

Step 1: Define the graph state and nodes

from typing import TypedDict, Literal
from langgraph.graph import StateGraph, END
from langchain_azure_ai.chat_models import AzureAIChatCompletionsModel
import os

class TicketState(TypedDict):
    user_input: str
    ticket_type: Literal["billing", "technical", ""]
    response: str

model = AzureAIChatCompletionsModel(
    endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
    model=os.environ["FOUNDRY_MODEL_NAME"],
)

def triage_node(state: TicketState) -> TicketState:
    result = model.invoke(
        f"Classify this support ticket as 'billing' or 'technical' only: {state['user_input']}"
    )
    ticket_type = "billing" if "billing" in result.content.lower() else "technical"
    return {**state, "ticket_type": ticket_type}

def billing_node(state: TicketState) -> TicketState:
    result = model.invoke(f"You are a billing specialist. Resolve: {state['user_input']}")
    return {**state, "response": result.content}

def technical_node(state: TicketState) -> TicketState:
    result = model.invoke(f"You are a technical specialist. Resolve: {state['user_input']}")
    return {**state, "response": result.content}

Step 2: Wire the conditional routing

def route(state: TicketState) -> str:
    return state["ticket_type"]

graph = StateGraph(TicketState)
graph.add_node("triage", triage_node)
graph.add_node("billing", billing_node)
graph.add_node("technical", technical_node)
graph.set_entry_point("triage")
graph.add_conditional_edges("triage", route, {"billing": "billing", "technical": "technical"})
graph.add_edge("billing", END)
graph.add_edge("technical", END)

compiled_graph = graph.compile()

Notice the model calls point at FOUNDRY_PROJECT_ENDPOINT, not directly at Azure OpenAI or any single provider's SDK. This is what lets the same graph run locally against a Foundry project and later behind the hosted runtime without changing model-calling code.

Step 3: Expose the graph for Foundry hosting

Foundry's langchain_azure_ai.agents.hosting package wraps a compiled LangGraph graph so it can be deployed as a Hosted Agent, speaking either the Responses or Invocations protocol.

from langchain_azure_ai.agents.hosting import LangGraphResponsesHost

host = LangGraphResponsesHost(graph=compiled_graph)
app = host.create_app()

That app is what gets deployed — Foundry manages the runtime, session handling, and scaling from there. Locally you can run it with uvicorn app:app --reload and test over HTTP before deploying with azd ai agent deploy or the Foundry Toolkit extension for VS Code.

Local tools vs. built-in tools

One distinction that trips people up: LangGraph tools you attach in code run locally, colocated with your graph. Foundry's own built-in tools (file search, code interpreter, etc.) run server-side inside the Agent Service and don't add a node to your graph at all — they're invoked transparently when the service handles a request.

Aspect	Local (LangGraph) tools	Built-in Foundry tools
Execution location	Colocated with your graph process	Server-side in Foundry Agent Service
Graph representation	Adds a tool node	No node added
Applicable to	Any LangGraph/LangChain tool	Foundry agents only
Best for	Deterministic business logic, custom APIs	File search, code interpreter, managed retrieval

Tracing across two systems

Because the LangGraph portion runs outside Foundry's native agent runtime during local development, its traces don't automatically appear in the Foundry portal. You view LangGraph-level traces through Azure Monitor / Application Insights (Investigate → Agents), while Foundry-native agent traces show up directly in the Foundry portal. Once deployed as a Hosted Agent, LangChain and LangGraph applications can also be registered in the Foundry Control Plane for unified governance and trace visibility.

What this buys you in production

Once deployed, the same graph becomes available as a turnkey Application across Teams, M365 Copilot, and Bot Service channels without additional publishing code — Foundry handles the channel plumbing, while your LangGraph code stays focused purely on routing and reasoning logic.

DEV Community: Jubin Soni

A Deep Dive into the Microsoft Foundry Document Intelligence SDK: From PDF to Structured Data

The mental model first

Prerequisites

Step 1: layout extraction, straight to markdown

Step 2: pulling structured fields from a known document type

Step 3: add-on capabilities you'll want more often than the docs suggest

Step 4: build a classifier to route mixed document types

Step 5: build a custom extraction model for your own document type

Where this fits in the bigger picture

Production considerations before you commit

Where this leaves you

References

A Senior Engineer's Guide to Foundry IQ, MCP, and the OpenAI Agents SDK

The mental model first

Prerequisites

Step 1: point a plain OpenAI client at your Foundry deployment

Step 2: swap in the token provider and hand the client to the Agents SDK

Step 3: connect to the Knowledge Base over MCP

Step 4: build the agent and run it

Where the credentials actually live

Production considerations before you commit

Where this leaves you

References

Foundry IQ Auth, Explained: Managed Identity, OBO, and Everything Between

The mental model first

The four auth surfaces, up front

Prerequisites

Step 1: give a Search Index Knowledge Source something to point at

Step 2: upload a file directly, no storage account required

Step 3: federate a live external source over MCP

Step 4: assemble the Knowledge Base

Step 5: query it and watch it actually plan

Step 6: every KB is already an MCP server, use it directly

Step 7: ground a Foundry Agent

Step 8: or plug it into Microsoft Agent Framework instead

Surface 4: making retrieval actually respect who's asking

Surfaces 1 through 3, hardened: a checklist

Where this leaves you

References

Video and Audio as Knowledge Sources: Content Understanding into Microsoft Foundry IQ

The corpus you already have and never indexed

Architecture

The two paths, stated plainly

Prerequisites

Pin your API version before writing a line of code

Step 1 — Choose the analyzer, and know what each modality costs you

Step 2 — The fastest path from a recording to indexable text

Step 3 — Design a field schema worth extracting

Step 4 — Land the output where a knowledge source can reach it

Step 5 — Create the knowledge source and the knowledge base

What happens at query time

Making citations survive the pipeline

Troubleshooting

Where to go next

References

How to Connect a Foundry IQ Knowledge Base to LangGraph over MCP

Why this integration is worth doing

Architecture

Prerequisites

A word on API versions before you start

Step 1 — Create the knowledge source and knowledge base

Step 2 — Verify with the retrieve API before you touch MCP

Step 3 — Understand what the MCP endpoint gives you (and what it does not)

Step 4 — Authenticate, properly

The fix: an httpx.Auth that refreshes

Step 5 — Load the tool into LangGraph

Step 6 — Build the graph, and parse the response

Step 7 — Enforce per-user permissions

Step 8 — Decide who plans

Troubleshooting

Where to go next

References

Building Meeting Audio RAG on Microsoft Foundry with Fast Transcription and Foundry IQ

What we are building, and the shape of the data flow

Standing up the Foundry project and the Speech resource

Turning an hour of audio into diarized speaker turns

Making chunks that are worth citing

Wiring the transcripts into a Foundry IQ knowledge base

Giving the agent a knowledge base and a personality

The fix: an `httpx.Auth` that refreshes