DEV Community: Jubin Soni

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.

References

Implementing Zero-Trust Networking and Identity for Microsoft Foundry Agents

Jubin Soni — Sat, 04 Jul 2026 21:38:48 +0000

Once your multi-agent system (Parts 6-8) is functionally solid, the question that comes up in every enterprise security review is the same: how do you know an agent is only doing what it's authorized to do, over a network path you actually trust, with an audit trail that holds up when someone asks "what did this agent do and why" six months later? This post covers private networking, Entra Agent ID, and audit trail design.

Zero-trust network boundary

Private endpoints: keeping traffic off the public internet

By default, calls from your application to Foundry Models travel over the public internet (encrypted, but publicly routable). For regulated workloads, route through Private Link instead:

resource foundryPrivateEndpoint 'Microsoft.Network/privateEndpoints@2023-04-01' = {
  name: 'pe-foundry-project'
  location: location
  properties: {
    subnet: {
      id: subnetId
    }
    privateLinkServiceConnections: [
      {
        name: 'foundry-connection'
        properties: {
          privateLinkServiceId: foundryResourceId
          groupIds: ['account']
        }
      }
    ]
  }
}

Pair this with a Managed VNET at the project level so that outbound calls from your compute (the orchestration layer running the agent code from Part 7) never leave the private network boundary. The failure mode to check for: a dependency library making an unexpected public call (a package doing telemetry callback, for instance) that bypasses your intended network boundary entirely — audit your dependencies' network behavior, not just your own code's.

Entra Agent ID: identity for autonomous agents, not just services

Traditional managed identity was designed for services calling APIs on a fixed schedule with predictable behavior. Entra Agent ID extends this model specifically for autonomous agents that make independent decisions — the identity model needs to answer not just "is this call authenticated" but "was this agent authorized to take this specific action in this specific context."

from azure.identity import DefaultAzureCredential

class AgentIdentityContext:
    def __init__(self, agent_id: str, allowed_actions: list[str]):
        self.agent_id = agent_id
        self.allowed_actions = allowed_actions
        self.credential = DefaultAzureCredential()

    def authorize_action(self, action: str, resource_scope: str) -> bool:
        if action not in self.allowed_actions:
            log_authorization_denial(self.agent_id, action, resource_scope)
            return False
        token = self.credential.get_token(resource_scope)
        return token is not None

def execute_agent_action(agent_context: AgentIdentityContext, action: str, resource_scope: str, fn):
    if not agent_context.authorize_action(action, resource_scope):
        raise PermissionError(f"Agent {agent_context.agent_id} not authorized for {action}")
    return fn()

The key design point: each agent in your multi-agent chain gets its own scoped identity with its own explicit allowed_actions list, rather than all agents sharing one broad service principal. This means a compromised or misbehaving refund agent can't accidentally (or maliciously, if prompt-injected) invoke actions scoped only to the fraud-check agent.

Security control	What it protects against	Enforced where
Private Link / Managed VNET	Traffic leaving the trusted network boundary	Network layer
Entra Agent ID scoped identity	An agent invoking actions outside its role	Identity/authorization layer, in code
Structured audit log with reasoning	Inability to explain why an action was taken	Application logging layer
Authorization enforced in code, not model claims	Prompt injection claiming false authorization	Application logic, never the model's own output

Data residency and customer-managed keys

Private networking and identity handle "who can reach this system and act as whom." A separate, equally common enterprise requirement is "where does this data physically live, and who holds the encryption keys" — data residency and customer-managed keys (CMK) address this and get missed by teams who've handled networking and identity but stop there.

For data residency, confirm which region your Foundry project, its underlying Azure AI Search index, and any storage backing Foundry IQ actually run in — these can silently default to a different region than your primary application if not explicitly configured, which is a real problem for workloads subject to data-sovereignty requirements (GDPR-adjacent obligations, government contracts with residency clauses, etc.):

resource foundryProject 'Microsoft.CognitiveServices/accounts/projects@2024-10-01' = {
  name: 'your-project'
  location: 'westeurope'  // must match your data residency requirement explicitly —
                            // don't rely on the default location of the parent resource group
  properties: {
    // ...
  }
}

For customer-managed keys, the default is Microsoft-managed encryption at rest, which is adequate for most workloads but insufficient for some regulated industries that require the ability to revoke access to data by rotating or deleting a key the organization itself controls:

resource foundryAccount 'Microsoft.CognitiveServices/accounts@2024-10-01' = {
  properties: {
    encryption: {
      keySource: 'Microsoft.KeyVault'
      keyVaultProperties: {
        keyVaultUri: keyVaultUri
        keyName: 'foundry-cmk'
      }
    }
  }
}

Neither of these is something you retrofit easily after data has already been written under the default configuration — data residency and CMK are decisions to make explicitly during initial provisioning of the Foundry project, not something to leave as "we'll configure that before the security review." If your compliance requirements are still being finalized when you provision, default to the more restrictive configuration (explicit region pinning, CMK) rather than the platform default, since loosening a restriction later is far easier than migrating already-written data into a stricter configuration retroactively.

Audit trails that actually hold up

A log line saying "agent called refund API" is not an audit trail — when someone asks "why did the agent decide to issue this refund," you need the reasoning captured, not just the action:

import json
import time

def log_agent_decision(agent_id, state, decision, reasoning, authorized_by):
    audit_entry = {
        "timestamp": time.time(),
        "agent_id": agent_id,
        "task_id": state.task_id,
        "decision": decision,
        "reasoning_summary": reasoning,   # the model's stated rationale, captured verbatim
        "state_snapshot": {
            "order_id": state.order_id,
            "fraud_check_result": state.fraud_check_result,
        },
        "authorized_by": authorized_by,   # which identity/policy allowed this
        "input_hash": hash_conversation_context(state),  # for reproducibility without storing PII long-term
    }
    write_to_audit_log(audit_entry)

Three things a real audit trail needs that a simple action log doesn't: the model's stated reasoning at the decision point (not just the outcome), the authorization context (which identity/policy permitted the action), and enough state snapshot to reconstruct why the decision made sense given the inputs — without necessarily storing full raw PII long-term, which is why hashing the input context is often the right tradeoff between auditability and data minimization.

Testing the security boundary, not just the happy path

Write tests that deliberately attempt to violate the boundaries you've set up:

Attempt to call an action outside an agent's allowed_actions list and confirm it's denied and logged.
Attempt a request that would resolve to a public endpoint and confirm the network boundary rejects it.
Simulate a prompt-injection attempt that tries to get an agent to claim authorization for an action it doesn't have, and confirm the authorization check (not the model's own claim) is what gates execution.

That last point matters most: authorization must be enforced in code against the identity's actual permitted action list, never based on what the model says about its own permissions in its output text.

References

Private Link for Azure AI Foundry: https://learn.microsoft.com/en-us/azure/ai-foundry/how-to/configure-private-link
Managed VNET for Foundry projects: https://learn.microsoft.com/en-us/azure/ai-foundry/how-to/configure-managed-network
Microsoft Entra Agent ID: https://learn.microsoft.com/en-us/entra/identity/agent-identity/overview
Zero Trust architecture guidance: https://learn.microsoft.com/en-us/security/zero-trust/zero-trust-overview

Building Evaluation, Cost Governance, and Observability for a Multi-Agent System in Microsoft Foundry

Jubin Soni — Sat, 04 Jul 2026 21:33:37 +0000

This closes out the series' capstone: the multi-agent customer support system built across Parts 6-9, now hardened with evaluation, cost governance, and observability so it can actually run in production with an on-call rotation behind it, not just in a demo environment.

Continuous evaluation pipeline

Evaluation: measuring quality continuously, not just at launch

A one-time eval before launch tells you nothing about drift once real traffic — and real edge cases — start hitting the system. Set up a continuous evaluation pipeline using a G-Eval-style approach, where a separate model scores production outputs against explicit criteria:

eval_criteria = {
    "correctness": "Does the response accurately reflect the order/refund status retrieved from the tools?",
    "escalation_appropriateness": "If the case was ambiguous or high-risk, did the agent escalate to a human rather than resolving it alone?",
    "tone": "Is the response professional and appropriately empathetic given the customer's stated frustration level?",
}

def geval_score(response, context, criterion_name, criterion_description, eval_model_client):
    prompt = f"""Evaluate the following response against this criterion: {criterion_description}
Context: {context}
Response: {response}
Score from 1-5 and give one sentence of reasoning. Return JSON: {{"score": int, "reasoning": str}}"""
    result = eval_model_client.complete(prompt)
    return json.loads(result)

def run_continuous_eval(sample_of_production_traffic):
    scores = {crit: [] for crit in eval_criteria}
    for interaction in sample_of_production_traffic:
        for crit_name, crit_desc in eval_criteria.items():
            result = geval_score(interaction.response, interaction.context, crit_name, crit_desc, eval_model_client)
            scores[crit_name].append(result["score"])
    return {crit: sum(vals) / len(vals) for crit, vals in scores.items()}

Sample a percentage of real production traffic daily (not just synthetic test cases) and track these scores over time. A drop in escalation_appropriateness specifically is the metric most worth alerting on — it's a direct proxy for the system doing something risky without a human check, which is exactly the failure mode the recovery and authorization work in Parts 7 and 9 was designed to prevent.

Cost governance: PTU vs. pay-as-you-go, decided with real math

For a system with predictable, sustained traffic (which a production support system should have), Provisioned Throughput (PTU) usually beats pay-as-you-go on cost — but the crossover point depends on your actual volume:

def compare_ptu_vs_payg(monthly_token_volume, ptu_monthly_cost, payg_per_1k_tokens):
    payg_monthly_cost = (monthly_token_volume / 1000) * payg_per_1k_tokens
    return {
        "payg_monthly": payg_monthly_cost,
        "ptu_monthly": ptu_monthly_cost,
        "recommendation": "ptu" if ptu_monthly_cost < payg_monthly_cost else "payg",
        "breakeven_tokens": (ptu_monthly_cost / payg_per_1k_tokens) * 1000,
    }

Run this quarterly, not once — traffic volume for a maturing production system tends to grow, and the PTU crossover point is usually reached faster than teams expect once an agent system is handling a meaningful fraction of real support volume.

Chargeback tagging: attributing cost to the right owner

With multiple agents (fraud-check, refund, notification) potentially running on shared compute, tag at the project level so cost attribution doesn't require manual reconciliation later:

resource_tags = {
    "business-unit": "customer-support",
    "system": "multi-agent-refund-flow",
    "environment": "production",
    "cost-center": "CC-4471",
}

Apply these consistently at the Azure resource level (not just in application logs) so Cost Management reports can be filtered directly without a separate reconciliation step — this is the difference between a chargeback model that's usable monthly versus one that requires a data-engineering project every quarter.

Dashboard signal	Source	What it indicates
Request-level trace	Part 2 tracing patterns	Latency and failure location per agent step
Authorization denials	Part 9 identity logging	Potential security issue, not just a bug
Escalation rate vs. appropriateness score	Eval pipeline + agent logs	Whether the system is escalating correctly
Cost burn rate	Azure Cost Management tags	Budget overage risk before month-end

Observability: the on-call-ready dashboard

Pull together the tracing work from Part 2, the authorization logging from Part 9, and the eval scores above into a single dashboard an on-call engineer can actually use at 2am:

Request-level trace: which agents were invoked, in what order, with what latency per step (from Part 2's tracing patterns).
Authorization denials: any agent attempting an action outside its scope (from Part 9) — a spike here is a security signal, not just a bug signal.
Escalation rate: percentage of interactions escalated to a human, tracked against the eval-measured escalation_appropriateness score — a rising escalation rate paired with a falling appropriateness score means the system is escalating things it shouldn't, which is its own kind of problem.
Cost burn rate: token consumption against the PTU/PAYG budget, with an alert threshold before month-end overage becomes a surprise.

A concrete incident: what the on-call runbook actually looks like

All the observability infrastructure above is only as good as the runbook someone follows at 2am when an alert fires. Here's a worked example tying every prior post together into one incident response flow, using a realistic trigger: the escalation-rate alert from the dashboard fires, showing escalations up 3x over baseline in the last 30 minutes.

Step 1 — check the authorization denial log (Part 9). A spike in escalations correlated with a spike in authorization denials usually means an agent is attempting actions outside its scope — possibly a misconfigured deployment, possibly a prompt-injection attempt. This is checked first because it's the highest-severity possible cause.

Step 2 — check the circuit breaker state (Part 7). If a downstream dependency (the fraud-check API, say) is degraded, the circuit breaker should already be routing to human escalation rather than retrying — confirm it's open and working as designed, not that agents are timing out repeatedly without the breaker engaging.

Step 3 — check the eval scores for escalation_appropriateness (this post). If the score is stable and escalations are simply more frequent, this may be a legitimate traffic pattern (a genuinely higher-risk cohort of requests, e.g., during a known incident like a payment processor outage) rather than a system problem. If the score is dropping alongside the escalation spike, the system's judgment about when to escalate may itself be degrading — this points back toward Part 5's schema validation and Part 7's handoff logic as places to check for a recent regression.

Step 4 — check recent deployments against the canary process (Part 2). Cross-reference the timestamp of the spike against any recent flow, model version, or schema change. If a change went out in the last few hours without full canary ramp-up, that's the most likely single cause, and rollback is usually faster than root-causing forward.

def incident_triage(alert_context):
    checks = [
        ("authorization_denials", check_authorization_spike),
        ("circuit_breaker_state", check_circuit_breaker_status),
        ("eval_score_trend", check_escalation_appropriateness_trend),
        ("recent_deployments", check_recent_flow_changes),
    ]
    findings = {}
    for name, check_fn in checks:
        findings[name] = check_fn(alert_context)
        if findings[name].get("severity") == "critical":
            return {"triage_result": name, "findings": findings, "action": "immediate_rollback_or_escalation"}
    return {"triage_result": "inconclusive", "findings": findings, "action": "manual_investigation"}

Writing this ordering down explicitly — check security signals before assuming it's a quality regression, check for a bad deploy before deep root-causing — is what turns nine posts' worth of individually reasonable safeguards into something an on-call engineer who didn't build the system can actually execute under pressure.

References

Azure AI Foundry evaluation SDK: https://learn.microsoft.com/en-us/azure/ai-foundry/how-to/develop/evaluate-sdk
G-Eval and LLM-as-judge evaluation approach: https://learn.microsoft.com/en-us/azure/ai-foundry/concepts/observability
Provisioned Throughput Units (PTU) for Azure OpenAI: https://learn.microsoft.com/en-us/azure/ai-services/openai/concepts/provisioned-throughput
Azure Cost Management and tagging: https://learn.microsoft.com/en-us/azure/cost-management-billing/costs/cost-mgt-best-practices

Azure Databricks vs Microsoft Fabric: An Honest Guide to When to Use What

Jubin Soni — Mon, 29 Jun 2026 18:18:21 +0000

If you're building a data platform on Azure in 2026, you're going to be asked this question: Azure Databricks or Microsoft Fabric? Both run on Delta Lake, both integrate with ADLS Gen2, both have Spark, and both promise to be your unified data platform. The overlap is real and the marketing doesn't help.

This post is an honest breakdown of where each genuinely excels, where they overlap, and how to decide without getting lost in feature comparison tables.

Architecture Comparison

Decision Flow

Detailed Capability Comparison

Capability	Azure Databricks	Microsoft Fabric	Winner
Spark engine	Full Spark, Photon, tunable	Spark via Notebooks, less tunable	Databricks
Delta Lake	Native, full control	Via OneLake (Delta Parquet)	Tie
MLflow / MLOps	Native, full MLflow stack	Basic experiment tracking	Databricks
Model serving	Databricks Model Serving	Azure ML integration	Databricks
Power BI integration	DirectQuery via SQL Warehouse	Direct Lake (zero-copy, faster)	Fabric
SQL analytics	Serverless SQL Warehouse + Photon	SQL Analytics Endpoint	Tie
Data pipelines	Delta Live Tables, Workflows	Data Factory pipelines (mature)	Tie
Real-time intelligence	Spark Streaming + Kafka	Eventstream + KQL Database	Fabric
Setup complexity	Medium-high	Low (SaaS)	Fabric
Fine-grained governance	Unity Catalog (mature)	Purview integration (growing)	Databricks
Cost model	DBU + VM	Fabric capacity units	Comparable
Open format portability	High (standard Delta/Parquet)	Medium (OneLake but some lock-in)	Databricks

Step 1 — Reading Data from Fabric OneLake in Azure Databricks

The good news: Fabric and Databricks can share data via OneLake, which speaks Delta format. You don't have to pick one and abandon the other.

# Azure Databricks reading from Microsoft Fabric OneLake
# OneLake exposes an ABFS-compatible endpoint

# Authenticate using the workspace's Managed Identity or Service Principal
tenant_id     = dbutils.secrets.get("kv-scope", "sp-tenant-id")
client_id     = dbutils.secrets.get("kv-scope", "sp-client-id")
client_secret = dbutils.secrets.get("kv-scope", "sp-client-secret")

# OneLake uses the same ABFS protocol as ADLS Gen2
fabric_workspace_id = "your-fabric-workspace-guid"
lakehouse_name      = "your-lakehouse-name"
onelake_host        = "onelake.dfs.fabric.microsoft.com"

spark.conf.set(f"fs.azure.account.auth.type.{onelake_host}",             "OAuth")
spark.conf.set(f"fs.azure.account.oauth.provider.type.{onelake_host}",
               "org.apache.hadoop.fs.azurebfs.oauth2.ClientCredsTokenProvider")
spark.conf.set(f"fs.azure.account.oauth2.client.id.{onelake_host}",      client_id)
spark.conf.set(f"fs.azure.account.oauth2.client.secret.{onelake_host}",  client_secret)
spark.conf.set(f"fs.azure.account.oauth2.client.endpoint.{onelake_host}",
               f"https://login.microsoftonline.com/{tenant_id}/oauth2/token")

# Read a Delta table from Fabric Lakehouse
fabric_path = f"abfss://{fabric_workspace_id}@{onelake_host}/{lakehouse_name}.Lakehouse/Tables/sales_gold"

fabric_df = spark.read.format("delta").load(fabric_path)
print(f"Rows from Fabric Lakehouse: {fabric_df.count()}")
fabric_df.show(5)

Step 2 — Writing Databricks Results Back to OneLake

Run heavy ML feature engineering in Databricks, write results back to OneLake so Fabric Power BI can consume them via Direct Lake — zero-copy, sub-second dashboard refresh.

from pyspark.sql.functions import current_timestamp, lit

# Run your Databricks feature engineering / ML inference here
result_df = spark.table("production.gold.churn_predictions") \
    .withColumn("_computed_at", current_timestamp()) \
    .withColumn("_source",      lit("databricks-inference-job"))

# Write back to Fabric OneLake as Delta
output_path = f"abfss://{fabric_workspace_id}@{onelake_host}/{lakehouse_name}.Lakehouse/Tables/churn_predictions"

result_df.write \
    .format("delta") \
    .mode("overwrite") \
    .option("overwriteSchema", "true") \
    .save(output_path)

print(f"Written {result_df.count()} rows to Fabric OneLake.")
print("Power BI Direct Lake will pick up changes automatically.")

Step 3 — When to Use Fabric Notebooks vs Databricks Notebooks

Not everything needs Databricks. Fabric Notebooks are good enough for lighter data prep that feeds Power BI reports.

# This kind of transformation is fine in Fabric Notebooks
# Use Fabric when: output goes directly to Power BI, team is analytics-focused,
# no MLflow tracking needed, data volume < 100GB

# Fabric Notebook (PySpark — same syntax as Databricks)
from pyspark.sql.functions import col, sum as _sum, date_trunc

df = spark.read.format("delta").load("Tables/sales_silver")

summary = df \
    .withColumn("month", date_trunc("month", col("sale_ts"))) \
    .groupBy("month", "region", "product_category") \
    .agg(_sum("revenue").alias("monthly_revenue")) \
    .orderBy("month", "region")

# Write to Lakehouse table — Power BI picks it up via Direct Lake
summary.write.format("delta").mode("overwrite").saveAsTable("monthly_revenue_summary")

# Use Databricks when: MLflow tracking needed, complex ML pipeline,
# Unity Catalog governance required, data volume > 1TB, streaming workloads

When to Use Which: Decision Framework

# Use this as a mental checklist when deciding

DATABRICKS_STRENGTHS = [
    "Complex ML pipelines with MLflow experiment tracking",
    "Production model serving with A/B testing",
    "Fine-grained governance via Unity Catalog (row/column security)",
    "Spark Structured Streaming with Kafka / Event Hub",
    "Very large scale ETL (multi-TB, complex joins)",
    "Open-source tool integrations (dbt, Great Expectations, etc.)",
    "Multi-cloud or portability requirements",
]

FABRIC_STRENGTHS = [
    "Power BI as the primary consumption layer (Direct Lake = fastest)",
    "Analytics-focused teams without deep Spark expertise",
    "Microsoft 365 integration (Teams, SharePoint data sources)",
    "Real-time dashboards via Eventstream + KQL",
    "Fabric Data Factory for straightforward ELT pipelines",
    "Lower operational overhead — fully SaaS managed",
    "Already licensed via Microsoft 365 E5 / Fabric capacity",
]

BOTH_TOGETHER = [
    "Heavy ML/MLOps in Databricks, results published to OneLake for Power BI",
    "Fabric Data Factory for ingestion, Databricks for complex transformation",
    "Unity Catalog governing Databricks tables, Fabric consuming via shortcuts",
]

Things to Watch in Production

OneLake shortcuts are the integration bridge. Fabric Lakehouses support shortcuts that point to external Delta tables in ADLS Gen2 — the same storage Databricks writes to. This means Databricks writes once and Fabric reads without data movement. Set up shortcuts rather than copying data between platforms.

Unity Catalog doesn't govern Fabric. Your row-level security and column masks in Unity Catalog do not apply when Fabric reads the same underlying Delta files directly. If governance is critical, either run everything through Databricks or replicate governance rules in Fabric's permission model.

Fabric capacity units and Databricks DBUs are both usage-based but measure differently. Don't try to compare them directly. Run the same workload in both and compare wall-clock time and cost on your actual data sizes.

Fabric ML is improving fast but isn't MLflow. As of early 2026, Fabric ML experiment tracking is functional but doesn't have the depth of MLflow's model registry, artifact storage, or model serving. If MLOps maturity matters, stay on Databricks for ML.

Wrapping Up

The honest answer is: most mature Azure data platforms in 2026 use both. Azure Databricks for ML, complex transformations, governance, and streaming. Microsoft Fabric for Power BI-first analytics, simpler pipelines, and teams that don't need the full Databricks stack. OneLake shortcuts and the shared Delta format make them composable rather than competitive.

Pick based on your primary consumer: if it's Power BI dashboards, start with Fabric. If it's ML models and data products, start with Databricks. When you need both, they integrate cleanly.

References

MLOps: Building a CI/CD Pipeline for ML Models on Azure Databricks

Jubin Soni — Sun, 28 Jun 2026 15:03:49 +0000

Most ML teams are great at training models. Very few are great at shipping them. The gap between a notebook that works and a model that reliably serves production traffic is where most ML projects stall.

In this tutorial I'll walk through building a proper CI/CD pipeline for ML models on Azure Databricks using:

MLflow for experiment tracking, model versioning, and the model registry
Databricks Asset Bundles (DABs) for infrastructure-as-code and job deployment
Azure DevOps / GitHub Actions as the CI/CD orchestrator
Delta Lake as the feature and validation data store
Databricks Model Serving for the production endpoint

The use case is the same churn prediction model from the previous post, but this time we're focusing entirely on how it gets from a notebook to a production endpoint reliably and repeatably.

Architecture Overview

CI/CD Flow

Pipeline Stage Breakdown

Stage	Tool	What happens	Gate to next stage
CI	GitHub Actions	Lint, unit tests, bundle validate	All tests green
Train	Databricks Job	Full training run, MLflow logging	Job exits 0
Register	MLflow Registry	Model versioned and moved to Staging	Auto on train success
Validate	Databricks Job	Metric thresholds checked on holdout set	ROC-AUC >= 0.80
Promote	MLflow Registry	Model moved to Production	Validation passes
Deploy	Model Serving	Endpoint updated to new model version	Promotion complete
Monitor	Databricks Lakehouse Monitoring	Drift and accuracy tracked post-deploy	Ongoing

Step 1 — Project Structure with Databricks Asset Bundles

Databricks Asset Bundles (DABs) let you define your jobs, clusters, and pipelines as code and deploy them via CLI. This is your IaC layer.

# databricks.yml
bundle:
  name: churn-prediction

variables:
  env:
    default: dev

targets:
  dev:
    mode: development
    default: true
    workspace:
      host: https://adb-xxxx.azuredatabricks.net

  staging:
    workspace:
      host: https://adb-xxxx.azuredatabricks.net

  production:
    mode: production
    workspace:
      host: https://adb-xxxx.azuredatabricks.net

resources:
  jobs:
    training_job:
      name: churn-training-${var.env}
      tasks:
        - task_key: feature_engineering
          notebook_task:
            notebook_path: ./notebooks/01_feature_engineering.py
          new_cluster:
            spark_version: 14.3.x-scala2.12
            node_type_id: Standard_DS3_v2
            num_workers: 4

        - task_key: train_and_register
          depends_on:
            - task_key: feature_engineering
          notebook_task:
            notebook_path: ./notebooks/02_train_and_register.py
          new_cluster:
            spark_version: 14.3.x-scala2.12
            node_type_id: Standard_DS3_v2
            num_workers: 2

    validation_job:
      name: churn-validation-${var.env}
      tasks:
        - task_key: validate_model
          notebook_task:
            notebook_path: ./notebooks/03_validate_and_promote.py
          new_cluster:
            spark_version: 14.3.x-scala2.12
            node_type_id: Standard_DS3_v2
            num_workers: 2

Step 2 — Training Job with MLflow Autologging

Keep the training notebook clean and focused. Let MLflow autologging handle the heavy lifting on metric and param capture.

# notebooks/02_train_and_register.py
import mlflow
import mlflow.sklearn
from mlflow.models.signature import infer_signature
from sklearn.ensemble import GradientBoostingClassifier
from sklearn.model_selection import train_test_split
from sklearn.metrics import roc_auc_score, f1_score, precision_score, recall_score
from delta.tables import DeltaTable
import pandas as pd

mlflow.set_experiment('/churn-prediction/ci-cd-pipeline')
mlflow.sklearn.autolog(log_input_examples=True, log_model_signatures=True)

# Read Gold features and capture Delta version for reproducibility
gold_table    = DeltaTable.forName(spark, 'churn.gold.features')
delta_version = gold_table.history(1).select('version').collect()[0][0]

features_pdf = spark.table('churn.gold.features') \
    .filter("feature_date = current_date()") \
    .toPandas()

FEATURE_COLS = [
    'total_events', 'total_sessions', 'distinct_products',
    'events_last_30d', 'events_last_90d', 'days_since_last_event',
    'customer_tenure_days', 'avg_events_per_day',
    'recency_tier', 'engagement_score',
]
TARGET = 'churned'

X = features_pdf[FEATURE_COLS]
y = features_pdf[TARGET]
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42, stratify=y
)

with mlflow.start_run() as run:
    params = {
        'n_estimators':  200,
        'max_depth':     5,
        'learning_rate': 0.05,
        'subsample':     0.8,
    }
    model = GradientBoostingClassifier(**params, random_state=42)
    model.fit(X_train, y_train)

    y_prob = model.predict_proba(X_test)[:, 1]
    y_pred = model.predict(X_test)

    mlflow.log_param('delta_feature_version', delta_version)
    mlflow.log_param('git_sha', dbutils.widgets.get('git_sha'))  # passed in by CI
    mlflow.log_metric('roc_auc',   roc_auc_score(y_test, y_prob))
    mlflow.log_metric('f1_score',  f1_score(y_test, y_pred))
    mlflow.log_metric('precision', precision_score(y_test, y_pred))
    mlflow.log_metric('recall',    recall_score(y_test, y_pred))

    signature = infer_signature(X_train, y_pred)
    mlflow.sklearn.log_model(
        model,
        artifact_path='churn-gbm',
        signature=signature,
        registered_model_name='churn-prediction-gbm',
        await_registration_for=300,
    )

    # Pass run ID to downstream tasks via job output
    dbutils.jobs.taskValues.set(key='run_id', value=run.info.run_id)
    print(f"Training complete. Run ID: {run.info.run_id}")

Step 3 — Validation Job: Gate on Metrics Before Promoting

Never auto-promote based on a successful training run alone. Always validate on a held-out or recent dataset and check against a threshold before touching the Production alias.

# notebooks/03_validate_and_promote.py
import mlflow
from mlflow import MlflowClient
from sklearn.metrics import roc_auc_score
import pandas as pd

client    = MlflowClient()
MODEL_NAME = 'churn-prediction-gbm'

# Pick up run_id from the upstream training task
run_id = dbutils.jobs.taskValues.get(
    taskKey='train_and_register', key='run_id'
)

# Thresholds — fail the job if any metric is below these
THRESHOLDS = {
    'roc_auc':   0.80,
    'f1_score':  0.72,
    'precision': 0.70,
}

run     = client.get_run(run_id)
metrics = run.data.metrics

print("Validating metrics against thresholds...")
failures = []
for metric, threshold in THRESHOLDS.items():
    actual = metrics.get(metric, 0)
    status = 'PASS' if actual >= threshold else 'FAIL'
    print(f"  {metric}: {actual:.4f} (threshold: {threshold}) -> {status}")
    if actual < threshold:
        failures.append(f"{metric}={actual:.4f} below threshold {threshold}")

if failures:
    raise Exception(f"Validation failed: {', '.join(failures)}")

# Promote to Production alias if all thresholds pass
model_version = client.search_model_versions(
    filter_string=f"run_id='{run_id}'"
)[0].version

client.set_registered_model_alias(
    name=MODEL_NAME,
    alias='Production',
    version=model_version,
)

print(f"Model version {model_version} promoted to Production alias.")

Step 4 — GitHub Actions CI/CD Workflow

This is the glue that ties everything together. One workflow handles PR validation; the other handles deployment on merge to main.

# .github/workflows/mlops.yml
name: MLOps CI/CD

on:
  push:
    branches: [main]
  pull_request:
    branches: [main]

env:
  DATABRICKS_HOST:  ${{ secrets.DATABRICKS_HOST }}
  DATABRICKS_TOKEN: ${{ secrets.DATABRICKS_TOKEN }}

jobs:
  ci:
    name: Lint and Test
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.11'

      - name: Install dependencies
        run: pip install databricks-cli databricks-sdk pytest flake8 mlflow

      - name: Lint
        run: flake8 notebooks/ src/ --max-line-length=120

      - name: Unit tests
        run: pytest tests/unit/ -v

      - name: Validate DAB bundle
        run: databricks bundle validate --target staging

  train-and-deploy:
    name: Train, Validate, Deploy
    runs-on: ubuntu-latest
    needs: ci
    if: github.ref == 'refs/heads/main'
    steps:
      - uses: actions/checkout@v4

      - name: Install Databricks CLI
        run: pip install databricks-cli databricks-sdk

      - name: Deploy bundle to staging
        run: databricks bundle deploy --target staging

      - name: Run training job
        run: |
          JOB_RUN_ID=$(databricks bundle run training_job \
            --var "git_sha=${{ github.sha }}" \
            --output json | jq -r '.run_id')
          echo "TRAINING_RUN_ID=$JOB_RUN_ID" >> $GITHUB_ENV

      - name: Run validation job
        run: |
          databricks bundle run validation_job --target staging
          echo "Validation passed. Promoting to production."

      - name: Deploy bundle to production
        run: databricks bundle deploy --target production

      - name: Update Model Serving endpoint
        run: |
          python scripts/update_serving_endpoint.py \
            --model-name churn-prediction-gbm \
            --alias Production

Step 5 — Update the Serving Endpoint

The final step in the pipeline updates the Databricks Model Serving endpoint to point at the newly promoted Production model version.

# scripts/update_serving_endpoint.py
import argparse
from databricks.sdk import WorkspaceClient
from databricks.sdk.service.serving import ServedModelInput, EndpointCoreConfigInput
from mlflow import MlflowClient

def update_serving_endpoint(model_name: str, alias: str):
    w      = WorkspaceClient()
    client = MlflowClient()

    # Resolve alias to concrete version
    model_version = client.get_model_version_by_alias(model_name, alias).version
    print(f"Deploying {model_name} version {model_version} (alias: {alias})")

    endpoint_name = 'churn-prediction-endpoint'
    served_model  = ServedModelInput(
        model_name=model_name,
        model_version=model_version,
        workload_size='Small',
        scale_to_zero_enabled=True,
    )

    try:
        # Update existing endpoint
        w.serving_endpoints.update_config(
            name=endpoint_name,
            served_models=[served_model],
        )
        print(f"Endpoint '{endpoint_name}' updated to version {model_version}.")
    except Exception:
        # Create if it doesn't exist yet
        w.serving_endpoints.create(
            name=endpoint_name,
            config=EndpointCoreConfigInput(served_models=[served_model]),
        )
        print(f"Endpoint '{endpoint_name}' created with version {model_version}.")


if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('--model-name', required=True)
    parser.add_argument('--alias',      required=True)
    args = parser.parse_args()
    update_serving_endpoint(args.model_name, args.alias)

Tool Comparison

Tool	Role in pipeline	Why not the alternative
Databricks Asset Bundles	IaC for jobs and clusters	Terraform Databricks provider (more verbose, no native notebook support), manual UI (not reproducible)
MLflow Registry aliases	Production / Staging promotion	Stage-based promotion (deprecated in MLflow 2.x), manual version tracking (error prone)
GitHub Actions	CI/CD orchestrator	Azure DevOps (works equally well, swap yml syntax), Jenkins (more ops overhead)
Metric gate in validation job	Automated quality check	Manual review (blocks velocity), no gate at all (risky)
Databricks Model Serving	Managed REST endpoint	AKS deployment (more control, much more ops), Azure ML endpoints (extra service dependency)
dbutils.jobs.taskValues	Pass run_id between tasks	Environment variables (not available cross-task in DABs), hardcoded run lookup (fragile)

Things to Watch in Production

Pin your Databricks Runtime version. Using 14.3.x-scala2.12 in your bundle ensures every training run uses the same Spark and library versions. Floating versions (latest) cause silent library drift that breaks reproducibility.

Store secrets in Azure Key Vault, not GitHub Secrets alone. GitHub Secrets work fine for CI tokens but for long-lived service principal credentials that Databricks jobs use at runtime, back them with Azure Key Vault and reference them via Databricks secret scopes.

Set a metric baseline from your current production model. Your thresholds (ROC-AUC >= 0.80) should be relative to what the current Production model achieves on the same holdout set, not an arbitrary number. Add a step in the validation job that fetches the current Production model's metrics and gates the new model against those.

Tag every MLflow run with git SHA. Logging git_sha as a param in every training run means you can always trace a model artifact back to the exact code version that produced it. Critical for incident response.

Scale to zero on serving endpoints. For non-latency-critical models, enable scale_to_zero_enabled=True on your serving endpoint. It cuts cost dramatically for endpoints that don't receive traffic 24/7.

Wrapping Up

The pattern here is straightforward: code change triggers CI, CI triggers a training job, training job registers a model, a validation job gates on metrics, and only then does the model get promoted and deployed. Nothing manual, nothing skipped.

What makes this production-grade rather than just automated is the combination of Delta versioning for feature reproducibility, MLflow aliases for clean promotion semantics, and metric-gated promotion so a worse model can never silently replace a better one.

DEV Community: Jubin Soni

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

Failing well when retrieval or the model does not cooperate

Measuring whether the thing actually works

Getting this to production without regrets

How this compares to rolling the pipeline yourself

Where to take it next

References

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

How Lakebase gets there

Isolation strategies, compared

Building the loop

The full cycle

What branching does not solve

The broader shift

References

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

The shape of the problem

What Tool Search changes

Strategies, compared

Building it end to end

A turn, in sequence

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

Where it doesn't help

The general lesson

References

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

The problem with "just add more workers"

Architecture overview

Step 1 — Set up governed tables in Unity Catalog

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

Step 3 — The aggregation that triggers a shuffle

Step 4 — Fixing skew with salting

Step 5 — Write to Delta Lake with optimized writes

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

Comparing shuffle-mitigation techniques

Governance: tying it back to Unity Catalog

Production considerations

References

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

Why "search" isn't enough anymore

Architecture overview

Step 1 — Provision and index documents in Azure AI Search

Step 2 — Structured lookups in Cosmos DB

Step 3 — The agent loop with function calling

Step 4 — Session state in Cosmos DB

Comparing retrieval strategies

Production considerations

References

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

Architecture

Step 1: Create a knowledge source

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

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

Reasoning effort and cost/latency tradeoffs

Multi-hop queries in practice

References

The fix: an `httpx.Auth` that refreshes