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](../../../foundation/03-tools-and-data-access/SKILL.md). ## 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: ```python 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: ```bash 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](../../../foundation/03-tools-and-data-access/SKILL.md). --- ## 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](../../../foundation/05-knowledge-assistant/SKILL.md) — 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`](../../../../data_product_accelerator/skills/semantic-layer/03-genie-space-patterns/SKILL.md) and [`04-genie-space-export-import-api`](../../../../data_product_accelerator/skills/semantic-layer/04-genie-space-export-import-api/SKILL.md) — 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](../../../foundation/03-tools-and-data-access/SKILL.md) (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](../../../foundation/04-ai-gateway/SKILL.md) 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: ```json {"input": [{"role": "user", "content": "question"}]} ``` and parse the response as `output[].content[].text`: ```python 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: ```python 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`): ```yaml - 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: ```python 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 [