Use MCP tools in a Python agent

Connect to an MCP server, discover its tools, and run a Python agent that uses them. Use the URL from a Databricks-provided MCP, your registered MCP, or your server on Databricks Apps.

To use MCPs from Claude Code, Codex, or another coding agent, choose your client in Supported coding agents. For other assistants and MCP clients, see Other MCP clients.

For an Agent Bricks CLI project, add MCP tools with the CLI. The examples below show how to connect from your own Python code.

Prerequisites

  • Python 3.12 on your computer.
  • Your server's MCP URL. If someone shared an MCP with you, they must grant you access. Complete the provider login if it uses per-user OAuth. For a server on Databricks Apps, you need CAN USE on the app.
  • Access to a Azure Databricks model endpoint that supports tool calling. The example uses databricks-claude-sonnet-4-5. Replace it with an endpoint available in your workspace.

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.

For identity choices and connectivity requirements, see Authentication and network access.

Step 1: Install and sign in

  1. If you haven't installed the Databricks CLI, run this command on macOS or Linux:

    curl -fsSL https://raw.githubusercontent.com/databricks/setup-cli/main/install.sh | sh

    For Windows or other installation methods, see Install the Databricks CLI.

  2. Sign in to your workspace:

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

  3. Install the Python libraries:

    pip install --upgrade databricks-mcp databricks-sdk "mcp>=1.24,<2"

The examples use MCP Python 1.x, which is compatible with the agent frameworks below.

Step 2: Connect to your server

Save the following code as mcp_agent.py. Replace <mcp-server-url> with your server's URL:

  • Databricks-provided or registered MCP: https://<workspace-hostname>/ai-gateway/mcp-services/<catalog>.<schema>.<service>.
  • Server on Databricks Apps: Copy the app URL from its overview page and append /mcp.
from databricks_mcp import DatabricksMCPClient
from databricks.sdk import WorkspaceClient

workspace_client = WorkspaceClient()
server_url = "<mcp-server-url>"
mcp_client = DatabricksMCPClient(
    server_url=server_url,
    workspace_client=workspace_client,
)

tools = mcp_client.list_tools()
for tool in tools:
    print(tool.name, tool.description, tool.inputSchema, sep="\n")

Run the script:

python mcp_agent.py

You should see your server's tools, with descriptions and input schemas. Choose a read-only task that one of these tools supports for the next step. If the list is empty or the connection fails, see MCP authentication and networking.

Use another server

To find MCPs you can access in a catalog and schema, run:

databricks ai-gateway list-mcp-services --parent schemas/system.ai

Replace system.ai with your <catalog>.<schema> to list registered MCPs. The CLI handles pagination.

Step 3: Run an agent with these tools

Choose your framework, install its package, and append its Python example to mcp_agent.py. Each example uses the same server_url and workspace sign-in from step 2.

LangGraph

The example converts MCP content blocks to the model's chat message format with convert_to_openai_messages.

pip install --upgrade databricks-langchain langgraph

import asyncio
from databricks_langchain import (
    ChatDatabricks,
    DatabricksMCPServer,
    DatabricksMultiServerMCPClient,
)
from langchain_core.messages import convert_to_openai_messages
from langgraph.prebuilt import create_react_agent

async def main():
    client = DatabricksMultiServerMCPClient([
        DatabricksMCPServer(
            name="my-mcp-server",
            url=server_url,
            workspace_client=workspace_client,
        ),
    ])
    agent = create_react_agent(
        ChatDatabricks(endpoint="databricks-claude-sonnet-4-5"),
        tools=await client.get_tools(),
        prompt=lambda state: convert_to_openai_messages(state["messages"]),
    )
    task = input("Ask the agent to use a tool: ")
    result = await agent.ainvoke({
        "messages": [{"role": "user", "content": task}],
    })
    for message in result["messages"]:
        print(message)

asyncio.run(main())
Deployment notebook (optional)

For Model Serving deployment, adapt this notebook to use your server URL:

LangGraph MCP tool-calling agent

Get notebook

OpenAI Agents SDK

pip install --upgrade databricks-openai openai-agents

import asyncio
from agents import Agent, Runner, set_default_openai_api, set_default_openai_client
from agents.mcp import MCPServerStreamableHttpParams
from agents.tracing import set_trace_processors
from databricks_openai import AsyncDatabricksOpenAI
from databricks_openai.agents.mcp_server import McpServer

set_default_openai_client(AsyncDatabricksOpenAI())
set_default_openai_api("chat_completions")
set_trace_processors([])

async def main():
    async with McpServer(
        name="my-mcp-server",
        params=MCPServerStreamableHttpParams(url=server_url),
        workspace_client=workspace_client,
    ) as server:
        agent = Agent(
            name="Tool-using agent",
            instructions="Use the available tools to answer the user's question.",
            model="databricks-claude-sonnet-4-5",
            mcp_servers=[server],
        )
        task = input("Ask the agent to use a tool: ")
        result = await Runner.run(agent, task)
        for item in result.new_items:
            print(item.to_input_item())

asyncio.run(main())
Deployment notebook (optional)

For Model Serving deployment, adapt this notebook to use your server URL:

Agents SDK MCP tool-calling agent

Get notebook

OpenAI client

pip install --upgrade databricks-openai

This example runs the tool-calling loop explicitly using the OpenAI-compatible client.

import json
from databricks_openai import DatabricksOpenAI, McpServerToolkit

toolkit = McpServerToolkit(url=server_url, workspace_client=workspace_client)
tools_by_name = {tool.name: tool for tool in toolkit.get_tools()}
model_client = DatabricksOpenAI()
messages = [{"role": "user", "content": input("Ask the agent to use a tool: ")}]

for _ in range(10):
    response = model_client.chat.completions.create(
        model="databricks-claude-sonnet-4-5",
        messages=messages,
        tools=[tool.spec for tool in tools_by_name.values()],
    )
    message = response.choices[0].message
    messages.append(message.model_dump(exclude_none=True))
    if not message.tool_calls:
        print(message.content)
        break
    for call in message.tool_calls:
        try:
            tool = tools_by_name.get(call.function.name)
            if tool is None:
                raise ValueError(f"Unknown tool: {call.function.name}")
            arguments = json.loads(call.function.arguments or "{}")
            if not isinstance(arguments, dict):
                raise ValueError("Tool arguments must be a JSON object.")
            output = tool.execute(**arguments)
        except Exception as error:
            output = json.dumps({"error": str(error)})
        print(call.function.name, output)
        messages.append({
            "role": "tool",
            "tool_call_id": call.id,
            "content": str(output),
        })
else:
    raise RuntimeError("The agent reached the tool-calling limit.")
Deployment notebook (optional)

For Model Serving deployment, adapt this notebook to use your server URL:

OpenAI MCP tool-calling agent

Get notebook

Run python mcp_agent.py again. When prompted, ask for the read-only task you chose, including any required inputs. For example, if your server has a ticket search tool, ask it to find open tickets in a specific project.

Check the printed conversation for a tool call, its result, and the agent's answer. An answer without a tool call doesn't confirm that the MCP server was used.

Call a tool directly to troubleshoot

Use the tool name and input schema printed in step 2. This example prompts for the name and arguments so it works with your server's tools. Run it after the connection code in step 2:

import json

tool_name = input("Read-only tool name: ")
arguments = json.loads(input("Tool arguments as a JSON object: "))
result = mcp_client.call_tool(tool_name, arguments)
print(result)

Check that the result has no tool error and contains the data you expected.

Discover tool names and input schemas with list_tools() before calling a tool. Result formats vary by tool:

  • If structuredContent is present, use that structured result directly. A tool can describe its shape with outputSchema.
  • Otherwise, inspect the content blocks. Parse a text block as JSON only if the tool returns JSON. MCP also supports plain text and other content types.
  • Check isError and inspect a sample response before relying on particular output fields.

Deploy and share when you're ready

The local example runs as you. When you deploy the agent on Databricks Apps, choose the identity it uses: the app's service principal for shared access, or the calling user for per-user access.

For Databricks-provided or registered MCPs:

For legacy workspace servers or servers hosted on Databricks Apps, grant access to the underlying resources or app. See Agent authentication.

Additional resources

  • Last updated on