Create agent tools using Unity Catalog functions
Use Unity Catalog functions when your agent needs to run predefined logic, such as a SQL query whose parameters the agent supplies. For other tools, start with MCPs.
This guide uses Unity Catalog AI to create a small Python function, test it, and add it to an agent using your choice of framework. If you already have a function, install the libraries in step 1 and go to step 3.
Requirements
- A Databricks notebook with Python 3.10 or above. On classic compute, use Databricks Runtime 15.0 or above.
- Access to serverless compute for the identity that runs the code. If creating or running the function fails with a permission error, see Troubleshooting.
USE CATALOGandUSE SCHEMAon the function's parent catalog and schema. Creating a function also requiresCREATE FUNCTIONon the schema; calling an existing function requiresEXECUTEon it. See Function privileges.- A model that supports tool calling. The LangGraph example uses a Databricks model endpoint. The OpenAI, Anthropic, and LlamaIndex examples use provider API credentials, as described in their tabs.
Step 1: Install the libraries
Run the installation command for your framework in a notebook cell:
Framework | Installation command |
|---|---|
LangGraph |
|
OpenAI |
|
Anthropic |
|
LlamaIndex |
|
The LangGraph command pins a tested set of langchain, langgraph, and langgraph-prebuilt. Without the pins, the installer can pick a langgraph-prebuilt release that doesn't import with the installed langgraph.
For package source code, see the Unity Catalog AI repository.
To run the examples outside a notebook, install the same packages into a virtual environment, without the %:
python3 -m venv .venv
.venv/bin/pip install --upgrade <packages for your framework>
Then configure authentication for your workspace.
In a notebook, restart Python in a separate cell. Skip this step outside a notebook:
dbutils.library.restartPython()
Step 2: Create and test a function
Use DatabricksFunctionClient to create and test the function. Replace my_catalog and my_schema with an existing catalog and schema where you can create functions. This example creates add_numbers. replace=True overwrites an existing function with the same name, so you can rerun the example. Remove it if you don't want to overwrite a function.
from unitycatalog.ai.core.databricks import DatabricksFunctionClient
client = DatabricksFunctionClient()
CATALOG = "my_catalog"
SCHEMA = "my_schema"
def add_numbers(number_1: float, number_2: float) -> float:
"""Add two numbers.
Args:
number_1: The first number to add.
number_2: The second number to add.
Returns:
The sum of the two numbers.
"""
return number_1 + number_2
client.create_python_function(
func=add_numbers, catalog=CATALOG, schema=SCHEMA, replace=True
)
result = client.execute_function(
function_name=f"{CATALOG}.{SCHEMA}.add_numbers",
parameters={"number_1": 2.0, "number_2": 3.0},
)
if result.error:
raise RuntimeError(result.error)
print(result.value) # 5.0
Step 3: Use the function in your agent
Set function_name to your function's full name, then run the example for your framework. Each uses UCFunctionToolkit to call the function directly.
from unitycatalog.ai.core.databricks import DatabricksFunctionClient
function_name = "my_catalog.my_schema.add_numbers"
client = DatabricksFunctionClient()
question = "Use the tool to add 2.0 and 3.0."
- LangGraph
- OpenAI
- Anthropic
- LlamaIndex
This example uses the databricks-claude-sonnet-4-5 endpoint. Replace it with a tool-calling model available in your workspace.
from databricks_langchain import ChatDatabricks, UCFunctionToolkit
from langchain.agents import create_agent
toolkit = UCFunctionToolkit(function_names=[function_name], client=client)
agent = create_agent(
model=ChatDatabricks(endpoint="databricks-claude-sonnet-4-5"),
tools=toolkit.tools,
)
result = agent.invoke({"messages": [{"role": "user", "content": question}]})
print(result["messages"][-1].content)
See the LangChain integration reference.
Set the OPENAI_API_KEY and OPENAI_MODEL environment variables for your notebook process. Use a model that supports tool calling, and store the API key in a Databricks secret.
import os
from openai import OpenAI
from unitycatalog.ai.openai.toolkit import UCFunctionToolkit
from unitycatalog.ai.openai.utils import generate_tool_call_messages
toolkit = UCFunctionToolkit(function_names=[function_name], client=client)
model_client = OpenAI()
messages = [{"role": "user", "content": question}]
for _ in range(10):
response = model_client.chat.completions.create(
model=os.environ["OPENAI_MODEL"],
messages=messages,
tools=toolkit.tools,
)
if not response.choices[0].message.tool_calls:
print(response.choices[0].message.content)
break
messages.extend(generate_tool_call_messages(response=response, client=client))
else:
raise RuntimeError("The agent reached the tool-calling limit.")
The helper runs the requested functions and formats their results for the next model call. See the OpenAI integration reference.
Set the ANTHROPIC_API_KEY and ANTHROPIC_MODEL environment variables for your notebook process. Use a model that supports tool calling, and store the API key in a Databricks secret.
import os
from anthropic import Anthropic
from unitycatalog.ai.anthropic.toolkit import UCFunctionToolkit
from unitycatalog.ai.anthropic.utils import generate_tool_call_messages
toolkit = UCFunctionToolkit(function_names=[function_name], client=client)
model_client = Anthropic()
messages = [{"role": "user", "content": question}]
for _ in range(10):
response = model_client.messages.create(
model=os.environ["ANTHROPIC_MODEL"],
max_tokens=1024,
tools=toolkit.tools,
messages=messages,
)
if response.stop_reason != "tool_use":
print("\n".join(block.text for block in response.content if block.type == "text"))
break
messages.extend(
generate_tool_call_messages(
response=response, client=client, conversation_history=[],
)
)
else:
raise RuntimeError("The agent reached the tool-calling limit.")
The helper runs the requested functions. Append its messages to the conversation before the next model call. See the Anthropic integration reference.
This example uses an OpenAI model. Set the OPENAI_API_KEY and OPENAI_MODEL environment variables for your notebook process. Use a model that supports tool calling, and store the API key in a Databricks secret.
import os
from llama_index.core.agent.workflow import FunctionAgent
from llama_index.llms.openai import OpenAI
from unitycatalog.ai.llama_index.toolkit import UCFunctionToolkit
toolkit = UCFunctionToolkit(function_names=[function_name], client=client)
agent = FunctionAgent(
tools=toolkit.tools,
llm=OpenAI(model=os.environ["OPENAI_MODEL"]),
)
result = await agent.run(user_msg=question)
print(result)
Run the last two lines in a notebook cell that supports await. See the LlamaIndex integration reference.
The agent calls add_numbers and returns an answer containing 5. If you use an existing function, change question to a task that function supports.
For deployment and authentication, see Build a custom agent and Agent authentication.
Local development
For testing trusted Python functions, initialize the client with execution_mode="local" and pass it to your framework's toolkit:
client = DatabricksFunctionClient(execution_mode="local")
toolkit = UCFunctionToolkit(function_names=[function_name], client=client)
Local mode retrieves the function from Unity Catalog and runs it in a local subprocess. It does not support SQL functions. Use the default serverless mode for production, and run only trusted code. See Execution modes.
For timeouts, memory limits, and serverless settings, see the local execution limits and Function Client environment variables.
Use the workspace MCP endpoint
For existing MCP integrations, this LangGraph example connects to the same function through the workspace MCP endpoint. Enable the Managed MCP Servers preview and use an identity with access to the function.
Run %pip install --upgrade databricks-langchain "langchain==1.2.10" "langgraph==1.0.10" "langgraph-prebuilt==1.0.8" "mcp>=1.24,<2" in a notebook cell, then restart Python. Replace my_catalog/my_schema/add_numbers with your function's path.
from databricks.sdk import WorkspaceClient
from databricks_langchain import (
ChatDatabricks,
DatabricksMCPServer,
DatabricksMultiServerMCPClient,
)
from langchain.agents import create_agent
workspace_client = WorkspaceClient()
function_path = "my_catalog/my_schema/add_numbers"
mcp_client = DatabricksMultiServerMCPClient([
DatabricksMCPServer(
name="uc-functions",
url=f"{workspace_client.config.host}/api/2.0/mcp/functions/{function_path}",
workspace_client=workspace_client,
),
])
agent = create_agent(
model=ChatDatabricks(endpoint="databricks-claude-sonnet-4-5"),
tools=await mcp_client.get_tools(),
)
result = await agent.ainvoke({
"messages": [{"role": "user", "content": "Use the tool to add 2.0 and 3.0."}],
})
print(result["messages"][-1].content)
For deployment, see Agent authentication. For a function that calls an external API with http_request, see Legacy REST API tools.
Troubleshooting
-
PERMISSION_DENIED: Cannot access Spark Connectwhen you create or run a function. Your user or service principal can't use serverless compute in this workspace.DatabricksFunctionClientopens a serverless Spark Connect session to create functions and, by default, to run them. This applies in a notebook and outside a notebook through Databricks Connect. A SQL warehouse alone is not sufficient.To check serverless access before you start, run:
Pythonfrom databricks.connect import DatabricksSessionDatabricksSession.builder.serverless().getOrCreate().sql("SELECT 1").collect()Ask a workspace admin to give that identity access to serverless compute. Without that access, create the function with
CREATE FUNCTION ... LANGUAGE PYTHONin the SQL editor or on a SQL warehouse, and run it with the client in local mode.