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

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:

1. The MCP tool result has a different shape from the REST/SDK retrieve response.
2. Bearer tokens expire, and a static headers dict will fail an hour into a long-running graph.
3. Per-user permission filtering requires a second token, distinct from your service credential.
4. 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:

1. 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.
2. 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.
3. 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

1. 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.
2. Agentic retrieval in Azure AI Search — overview
3. Model Context Protocol (MCP) — LangChain docs — transports, custom httpx.Auth, session lifecycle.
4. langchain-mcp-adapters on GitHub
5. MultiServerMCPClient API reference
6. Foundry IQ: build smarter agents faster with unified knowledge and serverless retrieval — Build 2026 announcement; GA scope and the MCP server.
7. Foundry IQ: improve recall by up to 54% with knowledge bases — the constraint-preservation evaluation behind Step 8.