Build, Log and Deploy a Databricks Agent: ResponsesAgent to agents.deploy()

What this deep dive adds

Chapter 25 of the course shows the ResponsesAgent wrapper as a skeleton and logging as a list of steps. This post is the working version: a complete tool-calling agent.py, the items it returns, streaming, the driver notebook that logs, validates and registers it, and the agents.deploy() arguments.

Building agents with MLflow and the Agent Framework is a Section 3 (Application Development) objective in the 18 March 2026 exam guide; registering and serving are Section 4. How the deployed agent authenticates has its own deep dive: declared resources, on-behalf-of-user and secrets.

The complete agent.py

The agent lives in its own file so MLflow can log the code, not a pickle. It answers order questions with one tool, calling a Foundation Model API endpoint through the OpenAI-compatible client in databricks-openai.

import json
from typing import Generator
from uuid import uuid4

import mlflow
from mlflow.entities import SpanType
from mlflow.models import set_model
from mlflow.pyfunc import ResponsesAgent
from mlflow.types.responses import (
    ResponsesAgentRequest,
    ResponsesAgentResponse,
    ResponsesAgentStreamEvent,
    to_chat_completions_input,
)
from databricks_openai import DatabricksOpenAI

LLM_ENDPOINT = "databricks-claude-sonnet-4-5"
MAX_TURNS = 5
SYSTEM_PROMPT = (
    "You are a support agent for an online store. Use get_order_status "
    "for any question about a specific order. Never guess a status."
)

# Tool spec in the Chat Completions format the endpoint expects
TOOLS = [{
    "type": "function",
    "function": {
        "name": "get_order_status",
        "description": "Look up the shipping status of one order.",
        "parameters": {
            "type": "object",
            "properties": {"order_id": {"type": "string"}},
            "required": ["order_id"],
        },
    },
}]


@mlflow.trace(span_type=SpanType.TOOL)
def get_order_status(order_id: str) -> str:
    # Stand-in for a real lookup (a UC function, a table, an API)
    return json.dumps({"order_id": order_id, "status": "shipped"})


TOOL_FNS = {"get_order_status": get_order_status}


def done(item: dict) -> ResponsesAgentStreamEvent:
    # One finished output item: a message, a tool call or a tool result
    return ResponsesAgentStreamEvent(type="response.output_item.done", item=item)


class SupportAgent(ResponsesAgent):
    def __init__(self):
        # Auth comes from the notebook in dev, and from the
        # declared resources once deployed
        self.client = DatabricksOpenAI()

    @mlflow.trace(span_type=SpanType.LLM)
    def call_llm(self, messages: list[dict]) -> dict:
        resp = self.client.chat.completions.create(
            model=LLM_ENDPOINT, messages=messages, tools=TOOLS
        )
        return resp.choices[0].message.to_dict()

    def predict_stream(
        self, request: ResponsesAgentRequest
    ) -> Generator[ResponsesAgentStreamEvent, None, None]:
        messages = [{"role": "system", "content": SYSTEM_PROMPT}]
        messages += to_chat_completions_input(
            [item.model_dump() for item in request.input]
        )
        for _ in range(MAX_TURNS):
            msg = self.call_llm(messages)
            messages.append(msg)
            if not msg.get("tool_calls"):
                # The model answered: emit the text and end the loop
                yield done(self.create_text_output_item(
                    text=msg["content"], id=str(uuid4())))
                return
            for call in msg["tool_calls"]:
                name = call["function"]["name"]
                arguments = call["function"]["arguments"]
                yield done(self.create_function_call_item(
                    id=str(uuid4()), call_id=call["id"],
                    name=name, arguments=arguments))
                result = TOOL_FNS[name](**json.loads(arguments))
                messages.append({"role": "tool",
                                 "tool_call_id": call["id"],
                                 "content": result})
                yield done(self.create_function_call_output_item(
                    call_id=call["id"], output=result))
        yield done(self.create_text_output_item(
            text="I could not finish that request.", id=str(uuid4())))

    @mlflow.trace(span_type=SpanType.AGENT)
    def predict(self, request: ResponsesAgentRequest) -> ResponsesAgentResponse:
        items = [
            event.item
            for event in self.predict_stream(request)
            if event.type == "response.output_item.done"
        ]
        return ResponsesAgentResponse(output=items)


mlflow.openai.autolog()
AGENT = SupportAgent()
set_model(AGENT)

• The loop calls the LLM, runs any tool calls, appends each result as a tool message and asks again, until the LLM answers in text. MAX_TURNS stops a model that never stops calling tools.
• to_chat_completions_input turns the request's Responses-format items into Chat Completions messages, so multi-turn history works.
• The create_*_item helpers build output items in the Responses schema's shapes; MLflow says ResponsesAgentResponse and ResponsesAgentStreamEvent are the only objects to construct yourself.
• Tracing: mlflow.openai.autolog() records each LLM call and the @mlflow.trace decorators add agent and tool spans. set_model(AGENT) marks the object MLflow loads.

Built from the MLflow page "ResponsesAgent for Model Serving" and the Databricks page "Log and register agents (Model Serving)". Run against MLflow 3.18 and databricks-openai 0.17 with a stubbed LLM: one tool call, one tool result, one answer.

What the agent returns

The response is a list of output items, not one string, so AI Playground, evaluations and reviewers see the full path to the answer.

predict and predict_stream

predict returns the whole response; predict_stream yields events as they happen. Writing the loop once in predict_stream and collecting its finished items in predict is the pattern MLflow's examples use, so the two always agree.

To stream the answer token by token, yield self.create_text_delta(delta=..., item_id=...) events with one item_id, then a final response.output_item.done event with the complete text under that id. Databricks uses that final event for the trace and the AI Playground display.

The driver notebook: test, log, validate, register

A separate notebook drives the agent file (test it first with from agent import AGENT and AGENT.predict(...)). Log it as code, declaring its dependencies and every Databricks resource it calls:

import mlflow
from mlflow.models.resources import DatabricksServingEndpoint

with mlflow.start_run():
    logged = mlflow.pyfunc.log_model(
        name="agent",
        python_model="agent.py",
        input_example={"input": [
            {"role": "user", "content": "Where is order A123?"}]},
        pip_requirements=["mlflow", "databricks-openai"],
        resources=[DatabricksServingEndpoint(
            endpoint_name="databricks-claude-sonnet-4-5")],
    )

• python_model="agent.py" logs the file. At serving time Databricks rebuilds the environment and runs it to load the agent.
• No signature argument: for a ResponsesAgent, MLflow infers one from the request and response schemas and adds the metadata {"task": "agent/v1/responses"}. Other agents must pass a signature or an input example.
• resources lists what the deployed agent calls; Databricks issues short-lived credentials for exactly these. Anything left out fails only once the endpoint is live. Indexes use DatabricksVectorSearchIndex, Unity Catalog functions DatabricksFunction.

Before deploying, run the logged model in a fresh environment built from its requirements, so a missing package fails here rather than on an endpoint build:

mlflow.models.predict(
    model_uri=logged.model_uri,
    input_data={"input": [{"role": "user", "content": "Where is order A123?"}]},
    env_manager="uv",
)

Finally, register it as a version in Models in Unity Catalog, which agents.deploy() requires:

mlflow.set_registry_uri("databricks-uc")
UC_MODEL_NAME = "main.support.order_agent"
uc_model = mlflow.register_model(model_uri=logged.model_uri, name=UC_MODEL_NAME)

From the Databricks page "Log and register agents (Model Serving)" and the MLflow page "ResponsesAgent for Model Serving". MLflow 3 names the artifact with name; older examples use artifact_path.

agents.deploy(): arguments and what it sets up

from databricks import agents

deployment = agents.deploy(UC_MODEL_NAME, uc_model.version, scale_to_zero=True)
deployment.query_endpoint

Per the Databricks docs, the call also provisions credentials for the declared resources (after checking the endpoint owner's privileges), turns on real-time tracing to the active MLflow experiment, enables the Review App, and logs requests to AI Gateway inference tables. It can take up to 15 minutes.

Updating is the same call with a new version: it joins the existing endpoint and takes traffic once ready while the old version keeps serving. Change only the version to keep that zero-downtime. list_deployments(), get_deployments() and delete_deployment() manage what is running.

Arguments from the databricks-agents 1.13 deploy() signature; behavior from the Databricks page "Deploy an agent for AI applications (Model Serving)" (MLflow 3.1.3 or above).

What this means for the exam

Distractors swap in an older step: pickling the agent, the workspace registry with stages, or a hand-built endpoint.

FAQ