feat(agents): structured output

apps/dev-playground/server/agents/classifier/agent.ts

@@ -0,0 +1,22 @@
+import { createAgent } from "@databricks/appkit/beta";
+import { z } from "zod";
+
+// Structured-output demo: a tool-free agent whose answer is validated against
+// its `output` Zod schema instead of returned as text. Discovered from the
+// folder name (`classifier`). See docs/plugins/agents.md "Structured output"
+// for how the object surfaces on /chat, /invocations, and in-process runAgent.
+export default createAgent({
+  instructions:
+    "You are a support-ticket triage classifier. Read the user's message and " +
+    "classify it into one category, decide whether it is urgent (the user is " +
+    "blocked or reports an outage), and write a one-sentence summary.",
+  output: z.object({
+    category: z
+      .enum(["billing", "bug", "feature_request", "how_to", "other"])
+      .describe("The single best-fit category for the ticket."),
+    urgent: z
+      .boolean()
+      .describe("True only if the user is blocked or reports an outage."),
+    summary: z.string().describe("A one-sentence summary of the request."),
+  }),
+});

docs/docs/plugins/agents.md

@@ -313,6 +313,47 @@ const result = await runAgent(classifier, {
 
 MCP hosted tools (`mcpServer(...)`) still require `agents()` (they need a live MCP client). Supervisor-API hosted tools (`supervisorTools.*`), by contrast, **work in standalone `runAgent`** — the adapter has everything it needs to execute them server-side. This makes batch-eval / CI use of supervisor agents possible without `createApp`. Plugin tool dispatch in standalone mode runs as the service principal (no OBO) and **bypasses the agents-plugin approval gate** — treat standalone runAgent as a trusted-prompt environment (CI, batch eval, internal scripts), not as an exposed user-facing surface.
 
+## Structured output
+
+Give a code agent an `output` Zod schema and its final answer is validated against that schema instead of returned as freeform text:
+
+```ts
+import { createAgent } from "@databricks/appkit/beta";
+import { z } from "zod";
+
+export default createAgent({
+  instructions: "Classify the support ticket.",
+  output: z.object({
+    category: z.enum(["billing", "bug", "feature_request", "how_to", "other"]),
+    urgent: z.boolean(),
+    summary: z.string(),
+  }),
+});
+```
+
+The parsed, schema-valid object shows up on every non-streaming surface and on the stream:
+
+| Surface | Where the object appears |
+| --- | --- |
+| `POST /invocations`, `POST /responses` | top-level `output_parsed` field on the JSON envelope (next to `output`) |
+| `POST /chat` (SSE) | one final `appkit.structured_output` event (`{ data }`) **after** the streamed text |
+| in-process `runAgent` | `RunAgentResult.output`, statically typed as `z.infer` of the schema |
+
+```ts
+import { runAgent } from "@databricks/appkit/beta";
+
+const { output } = await runAgent(classifier, { messages: ticket });
+output?.category; // "billing" | "bug" | … | undefined  — fully typed
+```
+
+A per-call override is also available for one-off shapes: `runAgent(agent, input, { output: SomeSchema })` (it takes precedence over the agent's own schema and retypes the result).
+
+**How it works.** The agent answers normally first (streaming its visible text on `/chat`). Structured output is then produced by a dedicated **non-streaming** completion constrained by the schema via `response_format` — Databricks rejects `response_format` under streaming, so the structured call can't be the streamed one. The answer is validated as-is first (a cheap pre-check for a model that already emitted JSON); otherwise the non-streaming pass reformats it. Either way the JSON is validated with Zod; on a mismatch AppKit re-prompts with the validation errors (up to two retries, invisible to the `/chat` text stream) and then throws a `StructuredOutputError` carrying the last raw output — it never returns partial or unvalidated data. If an endpoint rejects `response_format` outright, AppKit strips it and retries, relying on the prompt plus Zod validation.
+
+:::note Scope
+Structured output is for **code-config agents** (`createAgent`). Markdown `agent.md` agents can't carry a schema, and sub-agents return text into their parent's context (typing buys nothing there). See the live API reference (`npx @databricks/appkit docs "appkit API reference"`) for the exact `createAgent` / `runAgent` / `StructuredOutputError` signatures.
+:::
+
 ## Adding agents to an existing app
 
 Already have an app and want to add agents? What you touch depends on the kind:

packages/appkit/src/agents/databricks.ts

@@ -7,6 +7,7 @@ import type {
 } from "shared";
 
 import {
+  query as servingQuery,
   type StreamBody,
   stream as servingStream,
 } from "../connectors/serving/client";
@@ -70,6 +71,51 @@ function applyGenerationParams(
   }
 }
 
+/**
+ * True when `err`'s message names structured output / `response_format`. Used
+ * to strip the param and retry when an endpoint rejects it — some endpoints
+ * don't support it at all, and Databricks returns the coarse
+ * `INVALID_PARAMETER_VALUE` ("Structured output is not currently supported…")
+ * with no granular code to key on. The Zod boundary in the resolver is the
+ * real guarantee, so a stray strip is harmless (validation still runs); that
+ * lets this stay a simple message test rather than probing status codes.
+ *
+ * ponytail: substring match — Databricks has no granular code for a
+ * response_format rejection; tighten to a code check if one ever ships.
+ */
+function isResponseFormatRejection(err: unknown): boolean {
+  const msg = err instanceof Error ? err.message : String(err);
+  return /response_format|json[_ ]schema|structured output/i.test(msg);
+}
+
+/** Pull the assistant message text out of a NON-streaming chat completion. */
+function extractMessageContent(parsed: unknown): string {
+  if (!isRecord(parsed)) return "";
+  const choices = parsed.choices;
+  if (!Array.isArray(choices) || choices.length === 0) return "";
+  const first = choices[0];
+  if (!isRecord(first)) return "";
+  const message = first.message;
+  if (!isRecord(message)) return "";
+  const content = message.content;
+  if (typeof content === "string") return content;
+  // Harmony/array content: concatenate the text parts.
+  if (Array.isArray(content)) {
+    let text = "";
+    for (const part of content) {
+      if (
+        isRecord(part) &&
+        part.type === "text" &&
+        typeof part.text === "string"
+      ) {
+        text += part.text;
+      }
+    }
+    return text;
+  }
+  return "";
+}
+
 function extractLlamaToolJsonSlice(text: string): string | undefined {
   const start = text.indexOf("[{");
   if (start < 0) return undefined;
@@ -118,6 +164,12 @@ function reasoningPartText(part: Record<string, unknown>): string {
   return text;
 }
 
+/** Non-streaming sibling of {@link StreamBody}; returns the parsed JSON body. */
+type QueryBody = (
+  body: Record<string, unknown>,
+  signal?: AbortSignal,
+) => Promise<unknown>;
+
 /**
  * Escape-hatch options: provide an `endpointUrl` + `authenticate()` and the
  * adapter uses a bare `fetch()` to call it. Useful for tests and for pointing
@@ -146,6 +198,8 @@ interface RawFetchAdapterOptions {
  */
 interface StreamBodyAdapterOptions {
   streamBody: StreamBody;
+  /** Non-streaming transport for structured-output completions. */
+  queryBody?: QueryBody;
   maxSteps?: number;
   maxTokens?: number;
   generationParams?: GenerationParams;
@@ -284,6 +338,8 @@ interface DeltaToolCall {
  */
 export class DatabricksAdapter implements AgentAdapter {
   private streamBody: StreamBody;
+  /** Non-streaming transport; present only when structured output is usable. */
+  private queryBody?: QueryBody;
   private maxSteps: number;
   private maxTokens: number;
   private generationParams: GenerationParams;
@@ -304,11 +360,13 @@ export class DatabricksAdapter implements AgentAdapter {
 
     if (isStreamBodyOptions(options)) {
       this.streamBody = options.streamBody;
+      this.queryBody = options.queryBody;
     } else {
       const { endpointUrl, authenticate } = options;
-      this.streamBody = async (body, signal) => {
-        const fetchSignal =
-          signal ?? AbortSignal.timeout(RAW_FETCH_DEFAULT_TIMEOUT_MS);
+      const doFetch = async (
+        body: Record<string, unknown>,
+        signal?: AbortSignal,
+      ): Promise<Response> => {
         const authHeaders = await authenticate();
         const response = await fetch(endpointUrl, {
           method: "POST",
@@ -318,17 +376,24 @@ export class DatabricksAdapter implements AgentAdapter {
             ...authHeaders,
           },
           body: JSON.stringify(body),
-          signal: fetchSignal,
+          signal: signal ?? AbortSignal.timeout(RAW_FETCH_DEFAULT_TIMEOUT_MS),
         });
         if (!response.ok) {
           const errorText = await response.text().catch(() => "Unknown error");
           throw new Error(
             `Databricks API error (${response.status}): ${errorText}`,
           );
         }
+        return response;
+      };
+      this.streamBody = async (body, signal) => {
+        const response = await doFetch(body, signal);
         if (!response.body) throw new Error("No response body");
         return response.body;
       };
+      // Non-streaming sibling for structured-output completions.
+      this.queryBody = async (body, signal) =>
+        (await doFetch({ ...body, stream: false }, signal)).json();
     }
   }
 
@@ -364,6 +429,13 @@ export class DatabricksAdapter implements AgentAdapter {
           body,
           signal,
         ),
+      queryBody: (body, signal) =>
+        servingQuery(
+          workspaceClient as unknown as Parameters<typeof servingQuery>[0],
+          endpointName,
+          body,
+          signal,
+        ),
       maxSteps,
       maxTokens,
       generationParams,
@@ -480,6 +552,27 @@ export class DatabricksAdapter implements AgentAdapter {
 
     yield { type: "status", status: "running" };
 
+    // Tool-free structuring pass (the resolver drives it via
+    // `run({ tools: [], outputSchema })`): a non-streaming, schema-constrained
+    // completion. The visible answer streams on the path below.
+    if (input.outputSchema && input.tools.length === 0) {
+      let text: string;
+      try {
+        text = await this.structuredCompletion(
+          messages,
+          input.outputSchema,
+          context,
+        );
+      } catch (err) {
+        const msg =
+          err instanceof Error ? err.message : "Structured request failed";
+        yield { type: "status", status: "error", error: msg };
+        throw err;
+      }
+      yield { type: "message", content: text };
+      return;
+    }
+
     for (let step = 0; step < this.maxSteps; step++) {
       if (context.signal?.aborted) break;
 
@@ -559,6 +652,48 @@ export class DatabricksAdapter implements AgentAdapter {
     }
   }
 
+  /**
+   * One tool-free, non-streaming completion constrained by `schema` via
+   * `response_format` (Databricks rejects `response_format` under
+   * `stream: true`). Returns the raw message text for the caller to validate;
+   * on a rejection, strips `response_format` and retries once (see
+   * {@link isResponseFormatRejection}).
+   */
+  private async structuredCompletion(
+    messages: OpenAIMessage[],
+    schema: Record<string, unknown>,
+    context: AgentRunContext,
+  ): Promise<string> {
+    if (!this.queryBody) {
+      throw new Error(
+        "DatabricksAdapter: structured output requires a non-streaming transport. " +
+          "Build the adapter via DatabricksAdapter.fromServingEndpoint / fromModelServing.",
+      );
+    }
+    const body: Record<string, unknown> = {
+      messages,
+      max_tokens: this.maxTokens,
+      response_format: {
+        type: "json_schema",
+        json_schema: { name: "structured_output", schema, strict: true },
+      },
+    };
+    applyGenerationParams(body, this.generationParams);
+
+    let parsed: unknown;
+    try {
+      parsed = await this.queryBody(body, context.signal);
+    } catch (err) {
+      if (isResponseFormatRejection(err)) {
+        delete body.response_format;
+        parsed = await this.queryBody(body, context.signal);
+      } else {
+        throw err;
+      }
+    }
+    return extractMessageContent(parsed);
+  }
+
   private async *streamCompletion(
     messages: OpenAIMessage[],
     tools: OpenAITool[],