Azure Databricks提供 system.ai.python_exec,这是一个内置的 Unity 目录函数,允许 AI 代理动态运行代理编写的Python代码、由用户提供或从代码库检索的代码。 它默认可用,可直接在 SQL 查询中使用:
SELECT python_exec("""
import random
numbers = [random.random() for _ in range(10)]
print(numbers)
""")
若要详细了解代理工具,请参阅 AI 代理工具。
将代码解释器添加到代理
若要将 python_exec 添加到您的代理中,请连接到 system.ai Unity Catalog 架构的托管 MCP 服务器。 代码解释器作为预配置的 MCP 工具提供。https://<workspace-hostname>/api/2.0/mcp/functions/system/ai/python_exec
OpenAI 代理 SDK (应用)
from agents import Agent, Runner
from databricks.sdk import WorkspaceClient
from databricks_openai.agents import McpServer
# WorkspaceClient picks up credentials from the environment (Databricks Apps, notebook, CLI)
workspace_client = WorkspaceClient()
host = workspace_client.config.host
# The context manager manages the MCP connection lifecycle and ensures cleanup on exit.
# from_uc_function constructs the endpoint URL from UC identifiers and wires in auth
# from workspace_client, avoiding hardcoded URLs and manual token handling.
async with McpServer.from_uc_function(
catalog="system",
schema="ai",
function_name="python_exec",
workspace_client=workspace_client,
name="code-interpreter",
) as code_interpreter:
agent = Agent(
name="Coding agent",
instructions="You are a helpful coding assistant. Use the python_exec tool to run code.",
model="databricks-claude-sonnet-4-5",
mcp_servers=[code_interpreter],
)
result = await Runner.run(agent, "Calculate the first 10 Fibonacci numbers")
print(result.final_output)
向应用授予对以下函数 databricks.yml的访问权限:
resources:
apps:
my_agent_app:
resources:
- name: 'python_exec'
uc_securable:
securable_full_name: 'system.ai.python_exec'
securable_type: 'FUNCTION'
permission: 'EXECUTE'
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
# DatabricksMultiServerMCPClient provides a unified get_tools() interface across
# multiple MCP servers, making it easy to add more servers later without refactoring.
mcp_client = DatabricksMultiServerMCPClient([
DatabricksMCPServer(
name="code-interpreter",
url=f"{host}/api/2.0/mcp/functions/system/ai/python_exec",
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": "Calculate the first 10 Fibonacci numbers"}]}
)
# LangGraph returns the full conversation history; the last message is the agent's final response
print(result["messages"][-1].content)
向应用授予对以下函数 databricks.yml的访问权限:
resources:
apps:
my_agent_app:
resources:
- name: 'python_exec'
uc_securable:
securable_full_name: 'system.ai.python_exec'
securable_type: 'FUNCTION'
permission: 'EXECUTE'
模型服务
from databricks.sdk import WorkspaceClient
from databricks_mcp import DatabricksMCPClient
import mlflow
workspace_client = WorkspaceClient()
host = workspace_client.config.host
mcp_client = DatabricksMCPClient(
server_url=f"{host}/api/2.0/mcp/functions/system/ai/python_exec",
workspace_client=workspace_client,
)
tools = mcp_client.list_tools()
# get_databricks_resources() extracts the UC permissions the agent needs at runtime.
# Passing these to log_model lets Model Serving grant access automatically at deployment,
# without requiring manual permission configuration.
mlflow.pyfunc.log_model(
"agent",
python_model=my_agent,
resources=mcp_client.get_databricks_resources(),
)
若要部署代理,请参阅部署用于生成 AI 应用程序的代理(模型服务)。 有关使用 MCP 资源的日志记录代理的详细信息,请参阅 使用 Databricks 托管 MCP 服务器。
后续步骤
- AI 代理工具 - 浏览代理的其他工具类型。
- 使用 Unity 目录函数创建 AI 代理工具 - 使用 Unity 目录函数创建自定义工具。
- 使用 Databricks 托管 MCP 服务器 - 详细了解 Azure Databricks 上的托管 MCP 服务器。