Use MCP servers in Custom Agents

Connect your agent code to any MCP server on Azure Databricks: Databricks-managed servers, external MCP servers registered as MCP Services, and custom servers hosted as Databricks apps. All of them expose the same MCP interface, so the agent code is the same. What differs is the server URL and how you authenticate.

The databricks-mcp Python library handles authentication to Azure Databricks MCP servers, so the same client code works across all three server types.

Get your server URL

Set up the MCP server first, then use its URL in the following examples:

Discover available MCP servers and tools

Before writing agent code, find which servers and tools you can use. Do not hardcode server names, tool names, or argument shapes from memory; confirm them from the workspace.

• Browse servers in the workspace. Go to AI Gateway > MCPs to see the MCP servers available to you. Azure Databricks provides ready-to-use built-in servers: managed MCP servers for your own Unity Catalog data and functions (Genie spaces, Vector Search indexes, and Unity Catalog functions), and built-in system.ai MCP Services for third-party SaaS tools such as Slack, GitHub, Google Drive, Google Calendar, Gmail, and Microsoft 365.

• List MCP Services programmatically. List the MCP Services in any catalog and schema with the Unity Catalog REST API. For example, the built-in services:

  databricks api get "/api/2.1/unity-catalog/mcp-services?parent=schemas/system.ai&page_size=100"
  

  Replace system.ai with your own <catalog>.<schema> to find services you registered. page_size is capped at 100, and the response includes a next_page_token when more services exist. To enumerate every service in a schema, repeat the request with page_token=<next_page_token> until the response returns no token:

  token=""
  while :; do
    page=$(databricks api get "/api/2.1/unity-catalog/mcp-services?parent=schemas/system.ai&page_size=100&page_token=$token")
    echo "$page"
    token=$(echo "$page" | jq -r '.next_page_token // empty')
    [ -z "$token" ] && break
  done
  

• List a server's tools from code. Point a DatabricksMCPClient at any server URL and call list_tools() to get each tool's name, description, and input schema at runtime, as shown in Connect and list tools. This is the reliable way to learn a server's exact tools and arguments.

Set up your environment

1. Use OAuth to authenticate to your workspace:

   databricks auth login --host https://<workspace-hostname>
   

2. When prompted, enter a profile name and note it for later. The default profile name is DEFAULT.

3. Verify you have a local environment with Python 3.12 or above, then install dependencies:

   pip install -U "mcp>=1.9" "databricks-sdk[openai]" "mlflow>=3.1.0" "databricks-agents>=1.0.0" "databricks-mcp"
   

   The agent-framework examples below need their own SDK. Add openai-agents databricks-openai for the OpenAI Agents SDK, or databricks-langchain langgraph for LangGraph.

Connect and list tools

Create a DatabricksMCPClient with the server URL and list its tools. The same client works for managed, external (MCP Service), and custom server URLs:

from databricks_mcp import DatabricksMCPClient
from databricks.sdk import WorkspaceClient

workspace_client = WorkspaceClient(profile="DEFAULT")
host = workspace_client.config.host

# Use a managed, MCP Service, or custom server URL:
mcp_server_url = f"{host}/api/2.0/mcp/functions/system/ai"

mcp_client = DatabricksMCPClient(server_url=mcp_server_url, workspace_client=workspace_client)
tools = mcp_client.list_tools()
print(f"Available tools: {[t.name for t in tools]}")

To call a tool directly:

result = mcp_client.call_tool("system__ai__python_exec", {"code": "print('Hello, world!')"})
print(result.content)

Authenticate

Select the authentication method that matches where your agent runs. For an external MCP Service, the caller must also have EXECUTE on the service. AI Gateway enforces this permission on every call.

Local environment

Authenticate to your workspace with OAuth (see Set up your environment) and pass the profile to the client:

workspace_client = WorkspaceClient(profile="DEFAULT")
mcp_client = DatabricksMCPClient(server_url=mcp_server_url, workspace_client=workspace_client)

Service principal

Use a service principal's OAuth credentials. Pass the values directly, or retrieve them from Azure Databricks secrets (for example, client_id=dbutils.secrets.get(scope="my-scope", key="client-id")):

workspace_client = WorkspaceClient(
    host="https://<workspace-hostname>",
    client_id="<client-id>",
    client_secret="<client-secret>",
)
mcp_client = DatabricksMCPClient(server_url=mcp_server_url, workspace_client=workspace_client)

When you log the agent, use DatabricksApps (custom) or the relevant resource as a resource. See Automatic authentication passthrough.

On-behalf-of-user

Use ModelServingUserCredentials so the agent acts with the calling user's permissions. See On-behalf-of-user authentication:

from databricks.sdk.credentials_provider import ModelServingUserCredentials

workspace_client = WorkspaceClient(credentials_strategy=ModelServingUserCredentials())
mcp_client = DatabricksMCPClient(server_url=mcp_server_url, workspace_client=workspace_client)

Log the agent model using the apps scope, and for managed servers include the corresponding OAuth scope for each server. See Available managed servers.

For an external or built-in MCP Service, instead add the ai-gateway user API scope (user_api_scopes: [ai-gateway]) and grant the calling user EXECUTE on the service. See Authenticate to MCP Services.

Build an agent

Use an agent framework to turn the MCP server's tools into an agent. Point the framework at the server URL and pass your authenticated WorkspaceClient.

OpenAI Agents SDK

import asyncio
from agents import Agent, Runner
from databricks.sdk import WorkspaceClient
from databricks_openai.agents import McpServer


async def main():
    workspace_client = WorkspaceClient()
    host = workspace_client.config.host

    async with McpServer(
        url=f"{host}/ai-gateway/mcp-services/main.default.github_mcp",
        name="github-mcp",
        workspace_client=workspace_client,
    ) as mcp_server:
        agent = Agent(
            name="Local agent",
            instructions="You are a helpful assistant with access to external services.",
            model="databricks-claude-sonnet-4-5",
            mcp_servers=[mcp_server],
        )
        result = await Runner.run(agent, "List my open GitHub pull requests.")
        print(result.final_output)


asyncio.run(main())

LangGraph

from databricks.sdk import WorkspaceClient
from databricks_langchain import ChatDatabricks, DatabricksMCPServer, DatabricksMultiServerMCPClient
from langgraph.prebuilt import create_react_agent

workspace_client = WorkspaceClient()
host = workspace_client.config.host

mcp_client = DatabricksMultiServerMCPClient([
    DatabricksMCPServer(
        name="external-service",
        url=f"{host}/ai-gateway/mcp-services/main.default.github_mcp",
        workspace_client=workspace_client,
    ),
])

async with mcp_client:
    tools = await mcp_client.get_tools()
    agent = create_react_agent(
        ChatDatabricks(endpoint="databricks-claude-sonnet-4-5"),
        tools=tools,
    )
    result = await agent.ainvoke(
        {"messages": [{"role": "user", "content": "List my open GitHub pull requests."}]}
    )
    print(result["messages"][-1].content)

MCP Python SDK

Build a framework-independent agent that discovers and calls tools across one or more MCP servers. Save the following as mcp_agent.py. It accepts a list of managed, MCP Service, and custom server URLs:

import json
import uuid
import asyncio
from typing import Any, Callable, List
from pydantic import BaseModel

import mlflow
from mlflow.pyfunc import ResponsesAgent
from mlflow.types.responses import ResponsesAgentRequest, ResponsesAgentResponse

from databricks_mcp import DatabricksMCPClient
from databricks.sdk import WorkspaceClient
from databricks_openai import DatabricksOpenAI

# 1) CONFIGURE YOUR ENDPOINTS/PROFILE
LLM_ENDPOINT_NAME = "databricks-claude-sonnet-4-5"
SYSTEM_PROMPT = "You are a helpful assistant."
DATABRICKS_CLI_PROFILE = "YOUR_DATABRICKS_CLI_PROFILE"
assert (
    DATABRICKS_CLI_PROFILE != "YOUR_DATABRICKS_CLI_PROFILE"
), "Set DATABRICKS_CLI_PROFILE to the Databricks CLI profile name you specified when configuring authentication to the workspace"
workspace_client = WorkspaceClient(profile=DATABRICKS_CLI_PROFILE)
host = workspace_client.config.host
# Add more server URLs here — managed, MCP Service, or custom:
MANAGED_MCP_SERVER_URLS = [
    f"{host}/api/2.0/mcp/functions/system/ai",
]
# Custom MCP servers hosted on Databricks apps, or MCP Service endpoints:
CUSTOM_MCP_SERVER_URLS = []


# 2) HELPER: convert between ResponsesAgent "message dict" and ChatCompletions format
def _to_chat_messages(msg: dict[str, Any]) -> List[dict]:
    msg_type = msg.get("type")
    if msg_type == "function_call":
        return [
            {
                "role": "assistant",
                "content": None,
                "tool_calls": [
                    {
                        "id": msg["call_id"],
                        "type": "function",
                        "function": {
                            "name": msg["name"],
                            "arguments": msg["arguments"],
                        },
                    }
                ],
            }
        ]
    elif msg_type == "message" and isinstance(msg["content"], list):
        return [
            {
                "role": "assistant" if msg["role"] == "assistant" else msg["role"],
                "content": content["text"],
            }
            for content in msg["content"]
        ]
    elif msg_type == "function_call_output":
        return [
            {
                "role": "tool",
                "content": msg["output"],
                "tool_call_id": msg["tool_call_id"],
            }
        ]
    else:
        return [
            {
                k: v
                for k, v in msg.items()
                if k in ("role", "content", "name", "tool_calls", "tool_call_id")
            }
        ]


# 3) MCP SESSION + TOOL-INVOCATION LOGIC
def _make_exec_fn(server_url: str, tool_name: str, ws: WorkspaceClient) -> Callable[..., str]:
    def exec_fn(**kwargs):
        mcp_client = DatabricksMCPClient(server_url=server_url, workspace_client=ws)
        response = mcp_client.call_tool(tool_name, kwargs)
        return "".join([c.text for c in response.content])

    return exec_fn


class ToolInfo(BaseModel):
    name: str
    spec: dict
    exec_fn: Callable


def _fetch_tool_infos(ws: WorkspaceClient, server_url: str) -> List[ToolInfo]:
    print(f"Listing tools from MCP server {server_url}")
    infos: List[ToolInfo] = []
    mcp_client = DatabricksMCPClient(server_url=server_url, workspace_client=ws)
    mcp_tools = mcp_client.list_tools()
    for t in mcp_tools:
        schema = t.inputSchema.copy()
        if "properties" not in schema:
            schema["properties"] = {}
        spec = {
            "type": "function",
            "function": {
                "name": t.name,
                "description": t.description,
                "parameters": schema,
            },
        }
        infos.append(
            ToolInfo(name=t.name, spec=spec, exec_fn=_make_exec_fn(server_url, t.name, ws))
        )
    return infos


# 4) SINGLE-TURN AGENT CLASS
class SingleTurnMCPAgent(ResponsesAgent):
    def _call_llm(self, history: List[dict], ws: WorkspaceClient, tool_infos):
        client = DatabricksOpenAI()
        flat_msgs = []
        for msg in history:
            flat_msgs.extend(_to_chat_messages(msg))
        return client.chat.completions.create(
            model=LLM_ENDPOINT_NAME,
            messages=flat_msgs,
            tools=[ti.spec for ti in tool_infos],
        )

    def predict(self, request: ResponsesAgentRequest) -> ResponsesAgentResponse:
        ws = WorkspaceClient(profile=DATABRICKS_CLI_PROFILE)

        history: List[dict] = [{"role": "system", "content": SYSTEM_PROMPT}]
        for inp in request.input:
            history.append(inp.model_dump())

        tool_infos = [
            tool_info
            for mcp_server_url in (MANAGED_MCP_SERVER_URLS + CUSTOM_MCP_SERVER_URLS)
            for tool_info in _fetch_tool_infos(ws, mcp_server_url)
        ]
        tools_dict = {tool_info.name: tool_info for tool_info in tool_infos}
        llm_resp = self._call_llm(history, ws, tool_infos)
        raw_choice = llm_resp.choices[0].message.to_dict()
        raw_choice["id"] = uuid.uuid4().hex
        history.append(raw_choice)

        tool_calls = raw_choice.get("tool_calls") or []
        if tool_calls:
            fc = tool_calls[0]
            name = fc["function"]["name"]
            args = json.loads(fc["function"]["arguments"])
            try:
                tool_info = tools_dict[name]
                result = tool_info.exec_fn(**args)
            except Exception as e:
                result = f"Error invoking {name}: {e}"

            history.append(
                {
                    "type": "function_call_output",
                    "role": "tool",
                    "id": uuid.uuid4().hex,
                    "tool_call_id": fc["id"],
                    "output": result,
                }
            )

            followup = self._call_llm(history, ws, tool_infos=[]).choices[0].message.to_dict()
            followup["id"] = uuid.uuid4().hex
            assistant_text = followup.get("content", "")
            return ResponsesAgentResponse(
                output=[
                    {
                        "id": uuid.uuid4().hex,
                        "type": "message",
                        "role": "assistant",
                        "content": [{"type": "output_text", "text": assistant_text}],
                    }
                ],
                custom_outputs=request.custom_inputs,
            )

        assistant_text = raw_choice.get("content", "")
        return ResponsesAgentResponse(
            output=[
                {
                    "id": uuid.uuid4().hex,
                    "type": "message",
                    "role": "assistant",
                    "content": [{"type": "output_text", "text": assistant_text}],
                }
            ],
            custom_outputs=request.custom_inputs,
        )


mlflow.models.set_model(SingleTurnMCPAgent())

if __name__ == "__main__":
    req = ResponsesAgentRequest(
        input=[{"role": "user", "content": "What's the 100th Fibonacci number?"}]
    )
    resp = SingleTurnMCPAgent().predict(req)
    for item in resp.output:
        print(item)

Example notebooks

The following notebooks show how to build LangGraph and OpenAI agents that call MCP tools across managed, external, and custom MCP servers:

LangGraph MCP tool-calling agent

Get notebook

OpenAI MCP tool-calling agent

Get notebook

Agents SDK MCP tool-calling agent

Get notebook

Deploy your agent

Azure Databricks recommends deploying agents on Databricks Apps, which lets you fully manage the agent code, server configuration, and git-based versioning. Alternatively, deploy on Model Serving.

Whichever you select, grant the agent access to every resource its MCP servers depend on. For example, CAN_RUN on a Genie Agent or SELECT on an AI Search index.

Databricks Apps (recommended)

Declare each resource your agent uses, including the resources behind every MCP server, under resources.apps.<app>.resources in databricks.yml, then deploy the bundle to grant the app's service principal access. For example, for an agent that uses the managed Genie and AI Search servers:

resources:
  apps:
    my_agent_app:
      name: 'my-agent-app'
      source_code_path: ./
      resources:
        - name: 'llm'
          serving_endpoint:
            name: 'databricks-claude-sonnet-4-5'
            permission: 'CAN_QUERY'
        - name: 'genie_space'
          genie_space:
            space_id: '<genie-space-id>'
            permission: 'CAN_RUN'
        - name: 'vector_index'
          uc_securable:
            securable_full_name: '<catalog>.<schema>.<index-name>'
            securable_type: 'TABLE'
            permission: 'SELECT'

databricks bundle deploy
databricks bundle run my_agent_app

For the full authoring and deployment workflow, see Author an agent and deploy it on Databricks Apps. For all resource types and permission values, see Authentication for agents.

Model Serving

Log the agent with all the resources it needs at logging time, then deploy. See Deploy an agent for AI applications (Model Serving) and Authentication for Databricks resources. Azure Databricks recommends the databricks-mcp package to derive MCP server resources:

• For managed MCP servers, use databricks_mcp.DatabricksMCPClient().get_databricks_resources(<server_url>) to retrieve the resources the server needs.
• For a custom MCP server hosted on a Databricks app, include the app as a resource when you log the model.

For example, to deploy the agent defined in mcp_agent.py:

import os
from databricks.sdk import WorkspaceClient
from databricks import agents
import mlflow
from mlflow.models.resources import DatabricksFunction, DatabricksServingEndpoint, DatabricksVectorSearchIndex
from mcp_agent import LLM_ENDPOINT_NAME
from databricks_mcp import DatabricksMCPClient

databricks_cli_profile = "YOUR_DATABRICKS_CLI_PROFILE"
assert (
    databricks_cli_profile != "YOUR_DATABRICKS_CLI_PROFILE"
), "Set databricks_cli_profile to the Databricks CLI profile name you specified when configuring authentication to the workspace"
workspace_client = WorkspaceClient(profile=databricks_cli_profile)
host = workspace_client.config.host

current_user = workspace_client.current_user.me().user_name
mlflow.set_tracking_uri(f"databricks://{databricks_cli_profile}")
mlflow.set_registry_uri(f"databricks-uc://{databricks_cli_profile}")
mlflow.set_experiment(f"/Users/{current_user}/databricks_docs_example_mcp_agent")
os.environ["DATABRICKS_CONFIG_PROFILE"] = databricks_cli_profile

MANAGED_MCP_SERVER_URLS = [
    f"{host}/api/2.0/mcp/functions/system/ai",
]

here = os.path.dirname(os.path.abspath(__file__))
agent_script = os.path.join(here, "mcp_agent.py")
resources = [
    DatabricksServingEndpoint(endpoint_name=LLM_ENDPOINT_NAME),
    DatabricksFunction("system.ai.python_exec"),
    # Uncomment to include a custom MCP server hosted on a Databricks app:
    # DatabricksApp(app_name="app-name")
]

for mcp_server_url in MANAGED_MCP_SERVER_URLS:
    mcp_client = DatabricksMCPClient(server_url=mcp_server_url, workspace_client=workspace_client)
    resources.extend(mcp_client.get_databricks_resources())

with mlflow.start_run():
    logged_model_info = mlflow.pyfunc.log_model(
        artifact_path="mcp_agent",
        python_model=agent_script,
        resources=resources,
    )

UC_MODEL_NAME = "main.default.databricks_docs_mcp_agent"
registered_model = mlflow.register_model(logged_model_info.model_uri, UC_MODEL_NAME)

agents.deploy(
    model_name=UC_MODEL_NAME,
    model_version=registered_model.version,
)

Next steps

• Connect agents to third-party tools with MCP Services to govern external MCP servers in Unity Catalog.
• AI Search and Databricks SQL to configure managed server tool behavior with _meta parameters.
• Connect MCPs to AI assistants and coding agents to use MCP servers from Claude, Cursor, and other tools.
• Connect agents to external MCPs and tools for an overview of all approaches to connect agents to external services.