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 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 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 Databricks workspace. Consumer access is sufficient to call MCPs through Unity Gateway.
- Grant each user
EXECUTEon the MCPs, plusUSE CATALOGandUSE SCHEMAon their parent catalog and schema. See Share an MCP.
Configure your app's authentication
Choose the option that matches your app:
- Databricks Apps
- External identity provider
- Enable user authorization for the app. Your app uses the forwarded Databricks user token directly, without token exchange or an account federation policy.
- Add the user authorization scopes:
ai-gatewayfor tool calls andunity-catalogfor discovery and sign-in status APIs. - Grant each user CAN USE on the app.
If your app already has a 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:
- Ask an account admin to configure an account federation policy for your identity provider. The policy defines the accepted issuer, audience, and user mapping.
- Get a token for the signed-in user from your identity provider. The first code step exchanges it for a 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 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
- External identity provider
Databricks Apps supplies the user's 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.
If you already have a Databricks OAuth token for the user, use it directly. Otherwise, exchange the user's identity provider token for a 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
- Python
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.
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
- Python
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
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.stateisACTIVE, the user is signed in. Go to Call tools. - If the request returns
NOT_FOUND, or the state is anything other thanACTIVE, show a sign-in button for the MCP.
MCPs that don't use per-user OAuth, such as the 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 Databricks browser session. If the user isn't signed in to the workspace in that browser, 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 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
- Python
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)
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.