03-tools-and-mcp

name	03-tools-and-mcp
description	Use when wiring tools into an OpenAI Agents SDK agent: local Python function tools, DatabricksMCPClient connections, and MCPServerSse for MCP servers. Track A Step 3. Builds on shared concepts from F3 (Tools and Data Access).
license	Apache-2.0
clients	["ide_cli","genie_code"]
bundle_resource	none
deploy_verb	none
deploy_note	Tool + MCP wiring (local function tools, DatabricksMCPClient, MCPServerSse) — code, no bundle resource. Resolves identically on both clients; on Genie Code use its built-in tool surface for ad-hoc calls and runDatabricksCli for grants. See `skills/genie-code-environment`.
coverage	full
metadata	{"last_verified":"2026-06-05","volatility":"high","upstream_sources":[],"author":"prashanth-subrahmanyam","version":"3.1.0","domain":"genai-agents","pipeline_position":"A3","consumes":"customized_agent, agent_class, mcp_server_knowledge","produces":"agent_with_tools, mcp_connections, resource_grants","grounded_in":"docs.databricks.com/aws/en/generative-ai/agent-framework/author-agent, docs.databricks.com/aws/en/generative-ai/agent-framework/build-agent-tool, openai.github.io/openai-agents-python/tools/"}
fields_read	["agent.tools","agent.mcp_servers","agent.knowledge_base_backend","docs.agent_tool_plan.selected_tools","docs.agent_tool_plan.selected_mcp_servers"]

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:

Class	What it answers	Backing skill	Wire-in pattern
Knowledge Assistant (managed doc Q&A with citations)	"What does our policy say about X?"	F5: Knowledge Assistant Lifecycle — produces ka_endpoint_name	Call the KA Model Serving endpoint from a @function_tool (see Wiring KA as a function tool below)
Genie Space (NL → SQL on UC tables, conversational analytics)	"How many active customers signed up last week?"	data_product_accelerator/skills/semantic-layer/03-genie-space-patterns and 04-genie-space-export-import-api — produces genie_space_id	Call Genie via the Databricks SDK from a @function_tool, or wire as Genie MCP server (see F3)
Unity Catalog functions (deterministic Python or SQL functions, registered in UC)	"Compute next-tier-distance for this customer"	F3: Tools and Data Access (UC Functions MCP section)	Wire as UC Functions MCP server, or call directly via the SDK
Vector Search MCP (custom retrieval pipeline)	"Custom hybrid search with re-ranker"	F3 (Vector Search MCP)	MCPServerSse direct wiring (see below)
Local Python tools (business logic, math, formatting)	"Compute margin", "format response"	This skill — @function_tool (see below)	Decorator

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:

Endpoint kind	Wire via	Request shape	Notes
Knowledge Assistant	Raw requests.post to {host}/serving-endpoints/{name}/invocations	Responses API: {"input": [{"role": "user", "content": "..."}]}	Parse output[].content[].text. Do NOT use serving_endpoints.query(messages=...) — KA rejects messages and the SDK's typing surfaces it as 'dict' object has no attribute 'as_dict'.
Foundation Model API (Sonnet, llama-3.3, gpt-5-2)	databricks_openai.AsyncDatabricksOpenAI or OpenAI(base_url=...)	OpenAI Chat Completions: messages=[...]	Standard OpenAI-compatible client patterns apply.
Genie Space	SDK two-call sequence	(1) w.genie.start_conversation_and_wait(space_id=..., content=...) returns description + attachments[] (2) w.genie.get_message_query_result_by_attachment(space_id, conversation_id, message_id, attachment_id) returns rows	First call alone surfaces the SQL plan as prose, not the rows — always make both calls. See Wiring Genie as a function tool below.
AI Gateway endpoint	REST POST /api/2.0/serving-endpoints	external_model.databricks-model-serving shape	See F4: AI Gateway for the create-endpoint payload.

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 [