hulpprogramma voor Python code-interpreteragent

Azure Databricks biedt system.ai.python_exec, een ingebouwde Unity Catalog-functie waarmee AI-agents dynamisch Python code kunnen uitvoeren die is geschreven door de agent, geleverd door een gebruiker of worden opgehaald uit een codebasis. Deze is standaard beschikbaar en kan rechtstreeks worden gebruikt in een SQL-query:

SELECT python_exec("""
import random
numbers = [random.random() for _ in range(10)]
print(numbers)
""")

Zie AI-agenthulpprogramma'svoor meer informatie over agenthulpprogramma's.

De code-interpreter toevoegen aan uw agent

Als u wilt toevoegen python_exec aan uw agent, maakt u verbinding met de beheerde MCP-server voor het system.ai Unity Catalog-schema. De code-interpreter is beschikbaar als een vooraf geconfigureerd MCP-hulpprogramma op https://<workspace-hostname>/api/2.0/mcp/functions/system/ai/python_exec.

OpenAI Agents SDK (Apps)

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)

Geef de app toegang tot de functie in 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 (Apps)

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)

Geef de app toegang tot de functie in 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'

Model serveren

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(),
)

Zie Een agent implementeren voor generatieve AI-toepassingen (Model Serving) om de agent te implementeren. Zie Databricks managed MCP-servers gebruiken voor meer informatie over het registreren van agents met MCP-resources.

Volgende stappen

  • Last updated on