supervisor-api-background-mode

name	supervisor-api-background-mode
description	Enable Supervisor API background mode for long-running agent tasks. Use when: (1) Agent needs to run tasks longer than HTTP timeout limits, (2) User says 'background mode', 'long-running', 'supervisor api', (3) Converting from streaming to background polling pattern, (4) Agent needs resilience to connection drops during execution.

Supervisor API Background Mode

Prerequisites:

1. Run quickstart first (uv run quickstart) — it creates the MLflow experiment and .env file needed by the server.
2. Follow the supervisor-api skill to set up the Supervisor API with hosted tools and permissions. This skill extends that setup with background mode support.

Background mode submits the request asynchronously (background=True), polls for completion, and streams the result back to the frontend. Use this when agent tasks may exceed HTTP timeout limits (complex multi-tool workflows, large data analysis, etc.).

Before Starting

Use the AskUserQuestion tool to ask: "How often should the agent poll for background task completion?" with options:

• Every 2 seconds — Fast response times, good for interactive use
• Every 10 seconds — Balanced between responsiveness and API load
• Every 30 seconds — Lower API load, suitable for very long-running tasks

Use their answer to set POLL_INTERVAL in agent_server/utils.py.

Architecture

Chat UI ──POST /api/chat──> Express ──streamText()──> Python @stream()
                                                        |
                                                        +-- responses.create(background=True, stream=False)
                                                        |   (returns response_id immediately)
                                                        |
                                                        +-- poll every 2s: responses.retrieve(id)
                                                        |   skip items with status queued/incomplete/in_progress
                                                        |   yield completed items
                                                        |
                                                        +-- convert items to stream events
                                                        |   (chunk text into word-based deltas)
                                                        |
Chat UI <──SSE stream──── Express <──stream──------+

What Changes from the Base Supervisor API

Aspect	Base Supervisor API	Background Mode
responses.create()	stream=True or stream=False	background=True, stream=False
Response	Immediate result or SSE stream	Returns response_id immediately
Result retrieval	Direct from response	Poll responses.retrieve(id) every 2s
Streaming to frontend	Native SSE from API	Simulated streaming (chunked text deltas)
Client pattern	Single DatabricksOpenAI	Single AsyncDatabricksOpenAI with use_ai_gateway=True
MCP tools	Executed in single request	Multi-turn approval flow (see gotcha #3)
Timeout	HTTP request timeout	No timeout — polls until terminal status

Step 1: Add agent_server/utils.py

This replaces the base Supervisor API's simple responses.create() call with a polling loop and streaming conversion.

Uses a single AsyncDatabricksOpenAI client with use_ai_gateway=True for both responses.create() and responses.retrieve().

import asyncio
import logging
from typing import AsyncGenerator
from uuid import uuid4

from databricks.sdk import WorkspaceClient
from databricks_openai import AsyncDatabricksOpenAI
from mlflow.types.responses import ResponsesAgentRequest, ResponsesAgentStreamEvent

POLL_INTERVAL = 2.0  # seconds between polls
INITIAL_POLL_DELAY = 1.0  # delay before first poll

logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)


def get_session_id(request: ResponsesAgentRequest) -> str | None:
    if request.context and request.context.conversation_id:
        return request.context.conversation_id
    if request.custom_inputs and isinstance(request.custom_inputs, dict):
        return request.custom_inputs.get("session_id")
    return None


def create_supervisor_client(
    workspace_client: WorkspaceClient | None = None,
) -> AsyncDatabricksOpenAI:
    """
    Create an AsyncDatabricksOpenAI client routed through AI Gateway.
    use_ai_gateway=True automatically resolves the correct AI Gateway endpoint.
    """
    workspace_client = workspace_client or WorkspaceClient()
    client = AsyncDatabricksOpenAI(
        workspace_client=workspace_client,
        use_ai_gateway=True,
    )
    return client


 () -> :
    
    ECHOED_TYPES = {, }
    count = 
     item  request.:
        item_dict = item.model_dump()  (item, )  item
        role = item_dict.get()
        item_type = item_dict.get()
         role ==   item_type  ECHOED_TYPES:
            count += 
     count


  () -> AsyncGenerator[, ]:
    
    skip_items = _count_history_items(request)  request  
    seen_item_count = skip_items
     skip_items > :
        logger.info()
    poll_count = 
    response_id = response.

    logger.info(
        
        
    )

    
     response.status   (, ):
        logger.info(
            
            
        )
         item  (response.output  [])[skip_items:]:
            item_dict = (
                item.model_dump()  (item, )  item
            )
            item_status = item_dict.get(, )
             item_status  (, , ):
                
             item_dict
        

     asyncio.sleep(INITIAL_POLL_DELAY)

     :
        poll_count += 
        logger.info()

        :
            response =  client.responses.retrieve(response_id)
         Exception  e:
            logger.warning()
             asyncio.sleep(POLL_INTERVAL)
            

        status = response.status

        current_items = response.output  []
        new_items = (current_items) - seen_item_count
        logger.info(
            
            
        )

         new_items > :
             idx, item  (current_items[seen_item_count:]):
                item_dict = (
                    item.model_dump()  (item, )  item
                )
                item_status = item_dict.get(, )
                item_id = item_dict.get()
                item_type = item_dict.get()
                
                 item_status  (, , ):
                    logger.info(
                        
                        
                    )
                    
                  item_id  item_type != :
                    logger.info(
                        
                        
                    )
                    
                logger.info(
                    
                    
                )
                 item_dict
                seen_item_count += 
            :
                
                seen_item_count = (current_items)

         status == :
            logger.info(
                
                
            )
            
         status  (, ):
            error_msg = (
                (response, , )
                 
            )
            logger.error()
             RuntimeError()

        logger.info()
         asyncio.sleep(POLL_INTERVAL)


 () -> []:
    
    words = text.split()
    chunks = []
     i  (, (words), chunk_size):
        chunk = .join(words[i : i + chunk_size])
         i + chunk_size < (words):
            chunk += 
        chunks.append(chunk)
     chunks


 () -> [ResponsesAgentStreamEvent]:
    
    events = []
    item_type = item.get()
    item_id = item.get(, (uuid4()))

     item_type == :
        seq = 
         content_part  item.get(, []):
             content_part.get() == :
                text = content_part.get(, )
                 chunk  _chunk_text(text):
                    events.append(
                        ResponsesAgentStreamEvent(
                            =,
                            item_id=item_id,
                            content_index=seq,
                            delta=chunk,
                        )
                    )
                    seq += 
        events.append(
            ResponsesAgentStreamEvent(
                =,
                item=item,
            )
        )
    :
        
        events.append(
            ResponsesAgentStreamEvent(
                =,
                item=item,
            )
        )

     events

Step 2: Update agent_server/agent.py

Replace the base Supervisor API handlers with async background mode handlers. The key differences from the base skill:

• Use async handlers (required for polling)
• Pass background=True, stream=False to responses.create()
• Poll with poll_background_response() instead of reading the response directly
• Convert output items to stream events with output_item_to_stream_events()
• Pass request to poll_background_response() so it can skip echoed history items in multi-turn conversations

Include your TOOLS list from the supervisor-api skill's Step 2 if you have hosted tools.

import asyncio
import logging
from typing import AsyncGenerator

import mlflow
from databricks.sdk import WorkspaceClient
from mlflow.genai.agent_server import invoke, stream
from mlflow.types.responses import (
    ResponsesAgentRequest,
    ResponsesAgentResponse,
    ResponsesAgentStreamEvent,
)

from agent_server.utils import (
    create_supervisor_client,
    get_session_id,
    output_item_to_stream_events,
    poll_background_response,
)

mlflow.openai.autolog()
logging.getLogger("mlflow.utils.autologging_utils").setLevel(logging.ERROR)
logging.basicConfig(level=logging.INFO)

logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

MODEL = "databricks-claude-sonnet-4"
SYSTEM_INSTRUCTIONS = "You are a helpful assistant."
TOOLS = [...]  # Your hosted tools from supervisor-api Step 2


def build_input(request: ResponsesAgentRequest) -> list[dict]:
    return [i.model_dump() for i in request.input]


@invoke()
async def invoke_handler(
    request: ResponsesAgentRequest,
) -> ResponsesAgentResponse:
    if session_id := get_session_id(request):
        mlflow.update_current_trace(
            metadata={"mlflow.trace.session": session_id}
        )

    workspace_client = WorkspaceClient()
    client = create_supervisor_client(workspace_client)

    logger.info(f"[invoke] Submitting background request with model={MODEL}")
    response = await client.responses.create(
        model=MODEL,
        instructions=SYSTEM_INSTRUCTIONS,
        =build_input(request),
        tools=TOOLS,
        background=,
        stream=,
    )
    logger.info(
        
        
    )

    
    
    output_items = []
      item  poll_background_response(client, response, request):
        logger.info(
            
            
        )
        output_items.append(item)

    logger.info()
     ResponsesAgentResponse(output=output_items)



  () -> AsyncGenerator[ResponsesAgentStreamEvent, ]:
     session_id := get_session_id(request):
        mlflow.update_current_trace(
            metadata={: session_id}
        )

    workspace_client = WorkspaceClient()
    client = create_supervisor_client(workspace_client)

    logger.info()
    response =  client.responses.create(
        model=MODEL,
        instructions=SYSTEM_INSTRUCTIONS,
        =build_input(request),
        tools=TOOLS,
        background=,
        stream=,
    )
    logger.info(
        
        
    )

    
    
    
      item  poll_background_response(client, response, request):
        events = output_item_to_stream_events(item)
        logger.info(
            
            
        )
         event  events:
             event
             asyncio.sleep()  
    logger.info()

Key Gotchas

1. Incomplete items during in_progress

While the response status is in_progress, the Supervisor API may return output items that are not yet complete (their status field will be queued, incomplete, or in_progress). These partial items may have id: None and will cause Pydantic validation errors in ResponsesAgentStreamEvent and ResponsesAgentResponse. Always break at the first incomplete item to preserve ordering — items after an incomplete one may also be incomplete or out of order. They'll appear as completed on a later poll.

2. Simulated streaming for the frontend

The chat frontend expects SSE streaming events. Since background mode returns the full text at once, output_item_to_stream_events() chunks text into 1-word deltas and the stream handler adds a 10ms delay between yields to simulate a realistic streaming experience.

3. MCP server tools require a multi-turn approval flow

MCP server tools (uc_connection or app) require a multi-turn approval flow — see the supervisor-api skill for the full explanation and example input.

In background mode, when an MCP tool call requires approval, the response reaches completed status (not in_progress) with mcp_approval_request items in the output. This naturally ends the polling loop. The mcp_approval_request items are returned to the frontend for the user to approve.

The approval follow-up is itself a full background mode cycle: the frontend sends a new request (with the original input + mcp_approval_request + mcp_approval_response appended) using background=True, receives a new response ID, and polls again until the final completed response with the tool result and assistant message.

4. No timeout on polling

The polling loop runs indefinitely until a terminal status (completed, failed, cancelled). There is no max poll time — this is intentional for long-running background tasks. The frontend chat proxy also has no explicit timeout enforced in code.

Testing

Test background mode directly against the Supervisor API

# Get auth
export DATABRICKS_HOST=$(databricks auth env --profile <PROFILE> | grep DATABRICKS_HOST | cut -d= -f2)
export DATABRICKS_TOKEN=$(databricks auth env --profile <PROFILE> | grep DATABRICKS_TOKEN | cut -d= -f2)

# Submit background request with a Genie space tool
curl -s "${DATABRICKS_HOST}/ai-gateway/mlflow/v1/responses" \
  -H "Authorization: Bearer ${DATABRICKS_TOKEN}" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "<MODEL>",
    "input": [{"role": "user", "content": "What were the top 5 products by revenue last quarter?"}],
    "tools": [
      {
        "type": "genie_space",
        "genie_space": {
          "description": "Query sales and revenue data",
          "id": "<genie-space-id>"
        }
      }
    ],
    "background": true,
    "stream": false
  }'

# Poll (use the id from above)
curl -s "${DATABRICKS_HOST}/ai-gateway/mlflow/v1/responses/<RESPONSE_ID>" \
  -H "Authorization: Bearer ${DATABRICKS_TOKEN}"

Test locally via the agent server

uv run start-app --no-ui

# In another terminal:
curl -X POST http://localhost:8000/invocations \
  -H "Content-Type: application/json" \
  -H "x-forwarded-access-token: <YOUR_TOKEN>" \
  -d '{
    "input": [{"role": "user", "content": [{"type": "input_text", "text": "Hello"}]}]
  }'

Example Expected log output

INFO:agent_server.agent:[stream] Submitting background request with model=databricks-claude-sonnet-4
INFO:agent_server.agent:[stream] Background request submitted: id=resp_xxx, status=queued
INFO:agent_server.utils:[poll] Starting polling for response_id=resp_xxx, interval=2.0s
INFO:agent_server.utils:[poll] Poll #1: status=in_progress, total_items=0, new_items=0
INFO:agent_server.utils:[poll] Waiting 2.0s before next poll...
INFO:agent_server.utils:[poll] Poll #2: status=completed, total_items=1, new_items=1
INFO:agent_server.utils:[poll] Yielding item: type=message, id=msg_xxx
INFO:agent_server.utils:[poll] Response completed after 2 polls, 1 total items
INFO:agent_server.agent:[stream] Received item type=message, emitting N stream events
INFO:agent_server.agent:[stream] Complete