03 Tools And MCP

Track A Step 3: Agent SDK Tool Wiring

Wire tools into your OpenAI Agents SDK agent so it can query data, search documents, call functions, and connect to external services.

This step focuses on how to wire tools into the Agent class. For background on which MCP servers exist, how external MCP works, resource grants, and retriever schemas, see F3: Tools and Data Access.

Tool Plan Input Contract

For the Agents Accelerator, prefer docs/agent_tool_plan.yaml over inferred defaults. Wire only the tools listed in selected_tools[].

Tool families not selected are skipped and recorded in verification as skipped, not failed.

For SQL MCP, enforce the read-only guardrails from the plan before any smoke test. Reject generated SQL containing INSERT, UPDATE, DELETE, DROP, ALTER, CREATE, MERGE, or TRUNCATE when readonly: true.

Use docs/agent_tool_plan.yaml.verification.tool_smoke_tests[] as the source of smoke prompts. Every selected tool must produce at least one MLflow TOOL span.

OpenAI Agents SDK SQL MCP wiring example:

from agents.mcp import MCPServerSse

sql_mcp = MCPServerSse(
    url=f"{host}/api/2.0/mcp/sql",
    headers={"Authorization": f"Bearer {workspace_client.config.token}"},
)

When to Use

• Your agent is running (Steps 1-2) but only generates text — it needs to do things: query data, search documents, call APIs.
• You want to connect to Databricks managed MCP servers or external MCP servers from an OpenAI Agents SDK agent.
• You need to add local function tools for custom business logic.

Foundation Prerequisite Check

Foundation Step 3 (Tools and Data Access) should be understood before this step. Verify:

python3 -c "from databricks_mcp import DatabricksMCPClient; print('F3: OK')" 2>/dev/null || echo "F3: FAIL — pip install databricks-mcp"

If F3 fails, install databricks-mcp and review F3: Tools and Data Access.

Canonical Tool Catalog (pick what your agent needs)

Track A custom agents typically combine three classes of tools. Pick from this catalog based on the data your agent must reach:

Canonical default for the SkyLoyalty walkthrough: Knowledge Assistant

• Genie Space + a small set of @function_tool helpers. This pairing covers structured analytics (Genie), unstructured Q&A (KA), and bespoke business logic (functions) without a custom RAG stack.

Serving-endpoint API shape per endpoint kind

Different endpoint kinds use different request shapes. The most common bug in this skill's history is reaching for Chat Completions on every endpoint; KA uses Responses API and Genie uses a two-call SDK sequence. Use this table as the canonical wiring reference:

Wiring KA as a function tool (Track A pattern)

KA exposes a Model Serving endpoint, but KA serves the Responses API, not Chat Completions. A serving_endpoints.query(name=..., messages=[...]) call against a KA endpoint fails with Invalid request: 'messages' field is not supported (and may surface earlier as a Python 'dict' object has no attribute 'as_dict' typing trap inside the SDK shim).

Call the invocations URL directly with the Responses-API request body:

{"input": [{"role": "user", "content": "question"}]}

and parse the response as output[].content[].text:

import os
import requests
from agents import function_tool
from databricks.sdk import WorkspaceClient

KA_ENDPOINT = os.environ["KA_ENDPOINT_NAME"]  # captured in F5

@function_tool
def search_policy_docs(query: str) -> str:
    """Search policy documents and return cited answers.

    Args:
        query: Natural-language question about loyalty program policy.
    """
    w = WorkspaceClient()
    host = w.config.host.rstrip("/")
    url = f"{host}/serving-endpoints/{KA_ENDPOINT}/invocations"

    # KA uses the Responses API: `input`, NOT Chat Completions `messages`.
    body = {"input": [{"role": "user", "content": query}]}
    resp = requests.post(
        url,
        headers={"Authorization": f"Bearer {w.config.token}",
                 "Content-Type": "application/json"},
        json=body,
        timeout=60,
    )
    resp.raise_for_status()
    payload = resp.json()

    # Responses API: payload["output"][i]["content"][j]["text"]
    parts = []
    for item in payload.get("output", []):
        for chunk in item.get("content", []):
            text = chunk.get("text")
            if text:
                parts.append(text)
    return "\n".join(parts) or "(no answer)"

DON'T use w.serving_endpoints.query(name=..., messages=[...]) against a KA endpoint. KA does not accept Chat Completions messages. Use raw requests.post with {"input": [...]} against the invocations URL as shown above. (Foundation Model API endpoints — Sonnet, llama-3.3, gpt-5-2 — do accept Chat Completions and can use the OpenAI-compatible client.)

Declare the KA endpoint as a serving_endpoint resource (CAN_QUERY) in databricks.yml so the deployed agent has access — see the resource-kind reference below.

Wiring Genie as a function tool

Genie spaces require a two-call sequence via the SDK. The first call returns the planner description and message metadata; the rows live on the attachment and must be fetched in a second call. Tools that return only the first call's output deliver SQL-as-prose to the LLM, not data.

1. w.genie.start_conversation_and_wait(space_id=..., content=...) — starts the conversation, runs the planner, and returns a GenieMessage whose attachments[] carry the query metadata (each attachment has an attachment_id). The message's content/text is the description, not the rows.
2. w.genie.get_message_query_result_by_attachment(space_id=..., conversation_id=..., message_id=..., attachment_id=...) — pulls the actual SQL execution result (rows + columns) for the chosen attachment.

Combine both into a single tool so the Agents SDK gets both the description and the rows in one return value:

import os
from agents import function_tool
from databricks.sdk import WorkspaceClient

GENIE_SPACE_ID = os.environ["GENIE_SPACE_ID"]  # captured during semantic-layer setup

@function_tool
def query_loyalty_analytics(question: str) -> str:
    """Answer analytics questions over loyalty data using Genie.

    Args:
        question: Natural-language analytics question (e.g. "active members by tier").
    """
    w = WorkspaceClient()

    # CALL 1: start_conversation_and_wait — planner runs, returns description.
    msg = w.genie.start_conversation_and_wait(
        space_id=GENIE_SPACE_ID,
        content=question,
    )
    description = (msg.content or "").strip()
    if not msg.attachments:
        return description or "(no answer)"

    # CALL 2: get_message_query_result_by_attachment — fetches the rows.
    attachment_id = msg.attachments[0].attachment_id
    result = w.genie.get_message_query_result_by_attachment(
        space_id=GENIE_SPACE_ID,
        conversation_id=msg.conversation_id,
        message_id=msg.message_id,
        attachment_id=attachment_id,
    )

    # Render rows as a small markdown-ish table for the LLM.
    rows_text = "(no rows)"
    sm = getattr(result, "statement_response", None)
    if sm and getattr(sm, "result", None) and getattr(sm.result, "data_array", None):
        rows = sm.result.data_array
        cols = [c.name for c in (sm.manifest.schema.columns or [])] if sm.manifest else []
        header = " | ".join(cols)
        body = "\n".join(" | ".join(str(c) for c in row) for row in rows)
        rows_text = f"{header}\n{body}" if header else body

    return f"{description}\n\n{rows_text}"

DON'T stop after start_conversation_and_wait. The first call may return only a description (and zero attachments for refusals); always fetch get_message_query_result_by_attachment for each attachment whose rows you want to surface to the LLM.

Declare a genie_space resource in databricks.yml. Bundle bindings use name + space_id (not a bare id):

- name: genie-space
  genie_space:
    space_id: ${var.genie_space_id}
    permission: CAN_RUN

The outer name: is the resource binding (referenced by valueFrom: in app.yaml); the inner space_id: is the Genie Space identifier the platform looks up. Plus declare the underlying sql_warehouse the space uses.

Local Function Tools

OpenAI Agents SDK: @function_tool

The simplest way to give your agent capabilities — define Python functions and decorate them:

from agents import Agent, function_tool
from datetime import datetime

@function_tool
def get_current_time() -> str:
    """Get the current date and time in ISO format."""
    return datetime.now().isoformat()

@function_tool
def calculate_metrics(revenue: float, cost: float) -> dict:
    """Calculate profit margin and ROI from revenue and cost.

    Args:
        revenue: Total revenue in dollars.
        cost: Total cost in dollars.
    """
    profit = revenue - cost
    margin = (profit / revenue * 100) if revenue > 0 else 0
    return {"profit": profit, "margin_pct": round(margin, 2)}

@function_tool
def search_knowledge_base(query: str, max_results: int = 5) -> list[dict]:
    """Search the internal knowledge base for relevant documents.

    Args:
        query: Natural language search query.
        max_results: Maximum number of results to return (default 5).
    """
    from databricks.sdk import WorkspaceClient
    w = WorkspaceClient()
    results = w.vector_search_indexes.query_index(
        index_name="prod.docs.knowledge_index",
        columns=["content", "source", "score"],
        query_text=query,
        num_results=max_results,
    )
    return [
        {"content": r["content"], "source": r["source"], "score": r["score"]}
        for r in results.result.data_array
    ]

agent = Agent(
    name="analyst",
    instructions="You are a data analyst. Use tools to answer questions accurately.",
    model="databricks-claude-sonnet-4-6",
    tools=[get_current_time, calculate_metrics, search_knowledge_base],
)

Tool docstrings are critical. The LLM uses them to decide when and how to call each tool. Write clear descriptions with typed arguments.

Wiring MCP Servers with DatabricksMCPClient

The databricks-mcp library (covered in F3) is the recommended way to connect to managed and external MCP servers. For the OpenAI Agents SDK, you can wire MCP tools alongside local function tools.

Using MCPServerSse (Direct Wiring)

The OpenAI Agents SDK natively supports MCP via MCPServerSse:

from agents import Agent
from agents.mcp import MCPServerSse
from databricks.sdk import WorkspaceClient

ws = WorkspaceClient()
host = ws.config.host

ai_search_mcp = MCPServerSse(
    # AI Search (formerly Vector Search); legacy /mcp/vector-search/ prefix still works
    url=f"https://{host}/api/2.0/mcp/ai-search/prod/docs/knowledge_index",
    headers={"Authorization": f"Bearer {ws.config.token}"},
)

sql_mcp = MCPServerSse(
    url=f"https://{host}/api/2.0/mcp/sql",
    headers={"Authorization": f"Bearer {ws.config.token}"},
)

github_mcp = MCPServerSse(
    url=f"https://{host}/api/2.0/mcp/external/github_connection",
    headers={"Authorization": f"Bearer {ws.config.token}"},
)

agent = Agent(
    name="analyst",
    instructions="Use tools to search documents, query data, and check code.",
    model="databricks-claude-sonnet-4-6",
    mcp_servers=[ai_search_mcp, sql_mcp, github_mcp],
)

Combining Local Tools and MCP Servers

Mix @function_tool definitions with MCP servers:

agent = Agent(
    name="analyst",
    instructions="Use tools to answer questions. Use calculate_metrics for math.",
    model="databricks-claude-sonnet-4-6",
    tools=[calculate_metrics, get_current_time],
    mcp_servers=[ai_search_mcp, sql_mcp],
)

Environment-Aware Auth for MCP

Handle auth consistently across local dev and Databricks Apps:

import os
from databricks.sdk import WorkspaceClient

def get_mcp_headers():
    w = WorkspaceClient()
    return {"Authorization": f"Bearer {w.config.token}"}

def get_workspace_host():
    return os.environ.get("DATABRICKS_HOST", "").rstrip("/")

Use get_mcp_headers() when constructing MCPServerSse. Adjust if using OBO (see A4: Authentication).

Resource Grants for Deployment

When your agent uses MCP servers or other Databricks resources, you need to declare workspace resource attachments and UC grants.

For the complete grant matrix, app.yaml/databricks.yml examples, UC SQL patterns, MLflow resources declarations, and the get_databricks_resources() auto-discovery pattern, see F3: Resource Grants.

Resource-kind quick reference

When declaring resources in databricks.yml, the kind depends on what the resource is. Use uc_securable only for UC objects that have no dedicated top-level resource type (functions, tables, schemas, volumes, models). Warehouses, serving endpoints, and UC connections have their own top-level kinds — do not wrap them in uc_securable.

genie_space resource shape: The DAB binding takes the outer name: (the resource binding label, used by valueFrom: in app.yaml) plus the inner space_id: (the Genie Space identifier). It does not accept a bare id: like sql_warehouse does. Using id: silently fails validation in some CLI versions and leaves the app unable to resolve the resource at runtime. Always use name + space_id.

Model grants are derived from docs/agent_tool_plan.yaml.runtime_config.llm.

• When provider == "databricks", grant CAN_QUERY on runtime_config.llm.endpoint.
• When provider == "ai_gateway", do not create or configure Gateway. Verify the pre-provisioned endpoint and required permissions are documented, then leave provisioning to the optional Gateway hardening step.

Verify the bundle schema accepts each kind your agent needs before deploying:

Client note: IDE runs this in a terminal; Genie Code runs the databricks bundle … command via runDatabricksCli. See skills/genie-code-environment.

# Generic "does this kind exist in my CLI's schema?"
databricks bundle schema \
  | jq '.definitions | keys[]' \
  | grep -i 'AppResource_'

Expected output includes AppResource_SQL_WAREHOUSE, AppResource_SERVING_ENDPOINT, AppResource_UC_SECURABLE, AppResource_CONNECTION, AppResource_GENIE_SPACE, and (for CLI >= 0.295.0) AppResource_APP. If a needed kind is missing, upgrade the CLI.

Canonical agent-app app.yaml + databricks.yml (KA + Genie + LLM)

Most Track A agents need a triad of resources: an LLM endpoint, a KA endpoint, and a Genie Space (with its underlying SQL warehouse). The canonical declaration is:

# app.yaml — Agent App
command:
  - mlflow
  - genai
  - agent-server
  - serve
  - --module
  - agent
env:
  - name: LLM_ENDPOINT
    valueFrom: llm-endpoint
  - name: KA_ENDPOINT_NAME
    valueFrom: ka-endpoint
  - name: GENIE_SPACE_ID
    valueFrom: genie-space
  - name: WAREHOUSE_ID
    valueFrom: sql-warehouse

# databricks.yml — Agent App resource block
resources:
  apps:
    ${var.agent_app_name}:
      source_code_path: ./
      resources:
        - name: llm-endpoint
          serving_endpoint:
            name: databricks-claude-sonnet-4-6
            permission: CAN_QUERY
        - name: ka-endpoint
          serving_endpoint:
            name: ${var.ka_endpoint_name}
            permission: CAN_QUERY
        - name: genie-space
          genie_space:
            space_id: ${var.genie_space_id}
            permission: CAN_RUN
        - name: sql-warehouse
          sql_warehouse:
            id: ${var.warehouse_id}
            permission: CAN_USE

This is the resource shape the SkyLoyalty walkthrough (canonical) uses. The valueFrom: <resource-name> mapping in app.yaml resolves to the resource's URL or id at runtime — your agent reads os.environ["KA_ENDPOINT_NAME"] etc. without hardcoding workspace identifiers.

Custom Retriever Schemas

If your agent includes a custom retriever (not using AI Bridge tools like VectorSearchRetrieverTool), you must declare the retriever's output schema for AI Playground source links and evaluation judges.

For the complete pattern, see F3: Custom Retriever Schemas.

SDLC connection: The retriever schema declared here feeds directly into the SDLC evaluation pipeline. S4: Evaluation Runs uses doc_uri to run groundedness and relevance judges against your eval dataset's expected_retrieved_context.

Testing Tools Locally

Interactive test

uv run start-app
# Then in browser at http://localhost:8000:
# "What time is it?" -> should invoke get_current_time
# "Calculate margin for $1M revenue and $600K cost" -> should invoke calculate_metrics

Programmatic test

import asyncio
from agents import Runner

async def test_tools():
    result = await Runner.run(agent, "What time is it?")
    print(f"Output: {result.final_output}")
    print(f"Tools called: {[s.name for s in result.raw_responses]}")

asyncio.run(test_tools())

Verify in MLflow traces

After calling the agent, check MLflow traces for TOOL spans alongside AGENT and LLM spans. Each tool call should appear as a distinct span with:

• Input: the tool arguments
• Output: the tool return value
• Span type: TOOL

DO / DON'T

DO — Write detailed tool docstrings with typed args

@function_tool
def search_products(
    query: str,
    category: str = "all",
    max_results: int = 10,
) -> list[dict]:
    """Search the product catalog by keyword.

    Args:
        query: Natural language search query for products.
        category: Filter by product category. Use "all" for no filter.
        max_results: Maximum number of products to return.
    """
    ...

DON'T — Use vague docstrings or skip type hints

@function_tool
def search(q, n=10):
    """Search for stuff."""  # LLM won't know what or how to search
    ...

DO — Handle tool errors gracefully

@function_tool
def query_database(sql: str) -> str:
    """Execute a SQL query against the analytics warehouse."""
    try:
        result = execute_sql(sql)
        return format_results(result)
    except Exception as e:
        return f"Query failed: {str(e)}. Please try a simpler query."

DON'T — Let tools raise unhandled exceptions

@function_tool
def query_database(sql: str) -> str:
    """Execute a SQL query."""
    return execute_sql(sql)  # Exception crashes the agent loop

Common Issues

Validation Gate

All must pass before proceeding:

• At least one local @function_tool added and callable by the agent
• Agent correctly invokes the tool when prompted (visible in chat and traces)
• MCP connection configured for at least one Databricks service (if applicable)
• Resource grants identified for all tools that need workspace resources
• Tool calls appear as TOOL spans in MLflow traces
• Tool errors are handled gracefully (no unhandled exceptions)
• If using custom retrievers: mlflow.models.set_retriever_schema() called at module level
• KA tool uses Responses API — request body is {"input": [...]} against /serving-endpoints/{name}/invocations, NOT serving_endpoints.query(messages=[...])
• Genie tool issues both calls — start_conversation_and_wait followed by get_message_query_result_by_attachment for each attachment whose rows are returned to the LLM
• databricks.yml genie_space resource uses name + space_id (not bare id:)

Next Step

After passing this gate, load and execute tracks/A-custom-agent-apps/04-authentication/SKILL.md (A4: Authentication).

Notes to Carry Forward

Related Skills

References

• Build a tool for your agent
• MCP servers on Databricks
• Author an agent
• OpenAI Agents SDK — Tools
• OpenAI Agents SDK — MCP
• Add resources to a Databricks app

Version History