Connect your app to MCP tools

Build your own app on top of Model Context Protocol (MCP) servers, including the MCPs in the system.ai schema and MCPs you register. Your app calls tools as the signed-in user. Unity Gateway governance controls which MCPs and tools each user can call.

Use user authorization when your app runs on Databricks Apps. If your app signs users in with an external identity provider, use OAuth token federation to get a Azure Databricks token. This guide covers apps that serve multiple users. To build and test an agent locally with your own credentials, see Call MCP tools from a Python agent.

For MCPs that use per-user OAuth, users also authorize their external provider account. The following steps cover both getting the Azure Databricks token and completing provider sign-in.

Configure authentication and access

Before adding code, give users access to the MCPs and configure your app's authentication.

Give users access to the MCPs

These requirements apply to both authentication methods:

  • Add each user to the Azure Databricks workspace. Consumer access is sufficient to call MCPs through Unity Gateway.
  • Grant each user EXECUTE on the MCPs, plus USE CATALOG and USE SCHEMA on their parent catalog and schema. See Share an MCP.

Configure your app's authentication

Choose the option that matches your app:

Databricks Apps

  1. Enable user authorization for the app. Your app uses the forwarded Azure Databricks user token directly, without token exchange or an account federation policy.
  2. Add the user authorization scopes: ai-gateway for tool calls and unity-catalog for discovery and sign-in status APIs.
  3. Grant each user CAN USE on the app.

External identity provider

If your app already has a Azure Databricks OAuth token for the user, use it directly and skip federation setup and token exchange. The token needs the ai-gateway and unity-catalog scopes for the APIs in this guide. Otherwise, complete these steps:

  1. Ask an account admin to configure an account federation policy for your identity provider. The policy defines the accepted issuer, audience, and user mapping.
  2. Get a token for the signed-in user from your identity provider. The first code step exchanges it for a Azure Databricks OAuth token with all-apis, which covers the APIs used in this guide.

Your app's server must also be able to reach the workspace. If IP access lists are enabled, allow its outbound IP. See MCP networking.

To use system.ai.dbsql, system.ai.sandbox, or system.ai.web_search, an account admin must enable the Unity Gateway beta from the account console Previews page. See Manage account previews.

Set up your app

Keep the token and client specific to the signed-in user throughout these steps.

Get a Azure Databricks token for the user

Choose the option that matches how your app authenticates users. Both options give you databricks_token for the remaining steps.

Databricks Apps

Databricks Apps supplies the user's Azure Databricks token in the x-forwarded-access-token request header. Use that token directly. Your app doesn't need to perform token exchange.

Read the token within the current request. For example, in a Flask request handler:

from flask import request

databricks_token = request.headers.get("x-forwarded-access-token")
if not databricks_token:
    raise ValueError("A Databricks user token is required.")

For other frameworks, see Retrieve user authorization credentials.

External identity provider

If you already have a Azure Databricks OAuth token for the user, use it directly. Otherwise, exchange the user's identity provider token for a Azure Databricks OAuth token:

import requests

response = requests.post(
    "https://<workspace-hostname>/oidc/v1/token",
    data={
        "subject_token": idp_token,
        "subject_token_type": "urn:ietf:params:oauth:token-type:jwt",
        "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
        "scope": "all-apis",
    },
)
response.raise_for_status()
databricks_token = response.json()["access_token"]

Use databricks_token in the following steps. The exchange doesn't return a refresh token. Use expires_in from the response to track expiration, and repeat the exchange with a valid identity provider token as needed. See Token exchange requirements and limitations.

List MCPs

List the MCPs the user can access in a schema. For Databricks-provided MCPs, use schemas/system.ai. For registered MCPs, use the schema where they are registered, such as schemas/main.default.

Databricks SDK

Install the SDK:

pip install --upgrade databricks-sdk

Create a client with the user's token and list the MCPs:

from databricks.sdk import WorkspaceClient

workspace_client = WorkspaceClient(
    host="https://<workspace-hostname>",
    token=databricks_token,
    auth_type="pat",
)

for service in workspace_client.ai_gateway.list_mcp_services(parent="schemas/system.ai"):
    print(service.name)

The SDK setting auth_type="pat" selects the supplied token instead of the app's environment credentials. The supplied token is still the user's OAuth token. Create a WorkspaceClient for each user so that every call uses that user's token. The following steps reuse workspace_client. The SDK handles pagination for you.

Python

Call the REST API with the user's token and list the MCPs:

import requests

url = "https://<workspace-hostname>/api/2.1/unity-catalog/mcp-services"
headers = {"Authorization": f"Bearer {databricks_token}"}
params = {"parent": "schemas/system.ai"}

while True:
    response = requests.get(url, headers=headers, params=params)
    response.raise_for_status()
    body = response.json()
    for service in body.get("mcp_services", []):
        print(service["name"])
    if not body.get("next_page_token"):
        break
    params["page_token"] = body["next_page_token"]

See List MCP services.

Check the user's sign-in status

For MCPs that use per-user OAuth, such as system.ai.github, check whether the user is signed in to the external provider:

Databricks SDK

Get the user's credential for the MCP:

from databricks.sdk.errors import NotFound
from databricks.sdk.service.catalog import ProvisioningInfoState

try:
    credential = workspace_client.ai_gateway.get_mcp_service_user_mapped_credential(
        name="mcp-services/system.ai.github"
    )
    signed_in = (
        credential.provisioning_info is not None
        and credential.provisioning_info.state == ProvisioningInfoState.ACTIVE
    )
except NotFound:
    signed_in = False

Python

Call the REST API with the user's token:

import requests

response = requests.get(
    "https://<workspace-hostname>/api/2.1/unity-catalog/mcp-services/system.ai.github/user-credentials",
    headers={"Authorization": f"Bearer {databricks_token}"},
)
if response.status_code == 404:
    signed_in = False
else:
    response.raise_for_status()
    signed_in = response.json().get("provisioning_info", {}).get("state") == "ACTIVE"
  • If provisioning_info.state is ACTIVE, the user is signed in. Go to Call tools.
  • If the request returns NOT_FOUND, or the state is anything other than ACTIVE, show a sign-in button for the MCP.

MCPs that don't use per-user OAuth, such as the Azure Databricks data and tools MCPs, don't need this step. See Get the caller's credential.

Sign the user in to the provider

When the user clicks your sign-in button, open the MCP's sign-in page in a popup. The user signs in on that page and authorizes access on the provider's consent screen.

const signInUrl = `https://${workspaceHostname}/mcp-service-login?name=${encodeURIComponent(mcpName)}`;
window.open(signInUrl, 'mcp-sign-in', 'popup,width=600,height=700');

The sign-in page doesn't send a message back to your app. To detect when sign-in finishes, check the sign-in status again when the user returns to your app.

The sign-in page uses the user's Azure Databricks browser session. If the user isn't signed in to the workspace in that browser, Azure Databricks asks them to sign in first. The provider credential is saved for the user who is signed in on the page, so that user must be the same user your app acts as.

Call tools

Send MCP requests to the MCP's URL, https://<workspace-hostname>/ai-gateway/mcp-services/<catalog>.<schema>.<service>, with the user's Azure Databricks token.

Your app can call tools directly, as shown in this step, or give them to an agent that decides which tools to call. To run an agent, follow Step 3: Run an agent with these tools and pass the workspace_client you created with the user's token in List MCPs, so the agent calls tools as that user. The agent frameworks use version 1 of the MCP Python SDK, so install the packages listed on that page.

Databricks SDK

Install the packages:

pip install --upgrade databricks-mcp

Connect with the user's workspace_client from List MCPs, then list and call tools:

from databricks_mcp import DatabricksMCPClient

mcp_client = DatabricksMCPClient(
    server_url="https://<workspace-hostname>/ai-gateway/mcp-services/system.ai.github",
    workspace_client=workspace_client,
)

for tool in mcp_client.list_tools():
    print(f"{tool.name}: {tool.description}")

result = mcp_client.call_tool("<tool-name>", arguments={})
print(result)

Python

This example uses version 2 of the MCP Python SDK:

pip install --upgrade "mcp>=2"

Open a session with the user's token, then list and call tools:

import asyncio

import httpx2
from mcp import ClientSession
from mcp.client.streamable_http import streamable_http_client

MCP_URL = "https://<workspace-hostname>/ai-gateway/mcp-services/system.ai.github"


async def main(databricks_token):
    http_client = httpx2.AsyncClient(headers={"Authorization": f"Bearer {databricks_token}"})
    async with http_client, streamable_http_client(MCP_URL, http_client=http_client) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            result = await session.list_tools()
            for tool in result.tools:
                print(f"{tool.name}: {tool.description}")

            result = await session.call_tool("<tool-name>", arguments={})
            print(result)


asyncio.run(main(databricks_token))

Discover tool names and input schemas with list_tools() before you call a tool, and check isError on each result. For how to read results, see Call a tool directly to troubleshoot. If the tool list is empty or the connection fails, see MCP authentication.

Additional resources