From 082e3bb4eb28da5062d85879859e727434d8d8e7 Mon Sep 17 00:00:00 2001
From: Abeautifulsnow <lcprunstone@163.com>
Date: Fri, 12 Jun 2026 13:18:36 +0800
Subject: [PATCH] feat(ai): add agent tool calling loop

---
 .../src/components/editor/AiAssistant.vue     | 104 ++-
 apps/desktop/src/i18n/locales/en.ts           |   3 +
 apps/desktop/src/i18n/locales/zh-CN.ts        |   3 +
 apps/desktop/src/lib/ai.ts                    |  53 +-
 apps/desktop/src/lib/api.ts                   |   2 +
 apps/desktop/src/lib/http.ts                  |  52 ++
 apps/desktop/src/lib/tauri.ts                 |  25 +
 crates/dbx-core/src/agent_events.rs           |  79 +++
 crates/dbx-core/src/agent_loop.rs             | 280 ++++++++
 crates/dbx-core/src/agent_tools.rs            | 204 ++++++
 crates/dbx-core/src/ai.rs                     | 597 +++++++++++++++++-
 crates/dbx-core/src/db/postgres.rs            |   2 +
 crates/dbx-core/src/lib.rs                    |   3 +
 .../dbx-core/src/schema/providers/native.rs   |  10 +-
 crates/dbx-web/src/main.rs                    |   1 +
 crates/dbx-web/src/routes/ai.rs               |  82 +++
 src-tauri/src/commands/ai.rs                  |  42 ++
 src-tauri/src/lib.rs                          |   1 +
 18 files changed, 1491 insertions(+), 52 deletions(-)
 create mode 100644 crates/dbx-core/src/agent_events.rs
 create mode 100644 crates/dbx-core/src/agent_loop.rs
 create mode 100644 crates/dbx-core/src/agent_tools.rs

diff --git a/apps/desktop/src/components/editor/AiAssistant.vue b/apps/desktop/src/components/editor/AiAssistant.vue
index 72d3dc5a8f..a9231137f5 100644
--- a/apps/desktop/src/components/editor/AiAssistant.vue
+++ b/apps/desktop/src/components/editor/AiAssistant.vue
@@ -1,5 +1,5 @@
 <script setup lang="ts">
-import { computed, nextTick, onMounted, onUnmounted, ref } from "vue";
+import { computed, nextTick, onMounted, onUnmounted, ref, type Component } from "vue";
 import { uuid } from "@/lib/utils";
 import { useI18n } from "vue-i18n";
 import { translateBackendError } from "@/i18n/backend-errors";
@@ -16,7 +16,8 @@ import { connectionIconType } from "@/lib/connectionPresentation";
 import DatabaseIcon from "@/components/icons/DatabaseIcon.vue";
 import { useQueryStore } from "@/stores/queryStore";
 import { useToast } from "@/composables/useToast";
-import { buildAiContext, runAiStream, type AiAction } from "@/lib/ai";
+import { buildAiContext, runAgentStream, type AiAction } from "@/lib/ai";
+import type { AgentEvent } from "@/lib/tauri";
 import { buildAiAgentPlan } from "@/lib/aiAgentPlan";
 import { buildAiAgentStepItems, type AiAgentStepItem, type AiAgentStepTone } from "@/lib/aiAgentStepPresentation";
 import { createAiShikiCodeHighlighter, type AiCodeHighlighter } from "@/lib/aiCodeHighlighter";
@@ -90,7 +91,7 @@ const selectedMentions = ref<AiTableMention[]>([]);
 let mentionTimer: ReturnType<typeof setTimeout> | undefined;
 let mentionRequestId = 0;
 
-const actionButtons: { action: AiAction; icon: any; key: string }[] = [
+const actionButtons: { action: AiAction; icon: Component; key: string }[] = [
   { action: "generate", icon: Wand2, key: "ai.actions.generate" },
   { action: "explain", icon: HelpCircle, key: "ai.actions.explain" },
   { action: "optimize", icon: Zap, key: "ai.actions.optimize" },
@@ -199,8 +200,9 @@ async function changeConnection(connectionId: string) {
     if (tab) {
       queryStore.updateDatabase(tab.id, database);
     }
-  } catch (e: any) {
-    toast(t("connection.connectFailed", { message: translateBackendError(t, e?.message || String(e)) }), 5000);
+  } catch (e: unknown) {
+    const message = e instanceof Error ? e.message : String(e);
+    toast(t("connection.connectFailed", { message: translateBackendError(t, message) }), 5000);
   }
 }
 
@@ -251,8 +253,10 @@ function agentStepClass(tone: AiAgentStepTone): string {
 }
 
 function agentStepTitle(step: AiAgentStepItem): string {
-  if (!step.titleKey) return t(step.labelKey);
-  return t(step.titleKey, step.titleParams || {});
+  if (step.titleKey) return t(step.titleKey, step.titleParams || {});
+  const tool = step.titleParams?.tool;
+  if (tool) return `${t(step.labelKey)}: ${tool}`;
+  return t(step.labelKey);
 }
 
 function toggleReasoning(index: number) {
@@ -355,9 +359,10 @@ async function loadMentionCandidates(query: string) {
     mentionCache.value[key] = candidates.slice(0, 40);
     mentionCandidates.value = mentionCache.value[key];
     mentionSelectedIndex.value = 0;
-  } catch (e: any) {
+  } catch (e: unknown) {
     if (requestId !== mentionRequestId) return;
-    mentionError.value = translateBackendError(t, e?.message || String(e));
+    const message = e instanceof Error ? e.message : String(e);
+    mentionError.value = translateBackendError(t, message);
     mentionCandidates.value = [];
   } finally {
     if (requestId === mentionRequestId) mentionLoading.value = false;
@@ -484,6 +489,7 @@ async function send() {
   const assistantIdx = messages.value.length - 1;
   const sessionId = uuid();
   currentSessionId.value = sessionId;
+  const agentEvents: AgentEvent[] = [];
   try {
     const context = await buildAiContext(props.tab, props.connection, {
       mentionedTables,
@@ -492,7 +498,7 @@ async function send() {
       role: m.role,
       content: m.content,
     }));
-    await runAiStream(
+    await runAgentStream(
       {
         config: settings.aiConfig,
         action: activeAction.value,
@@ -501,29 +507,65 @@ async function send() {
         context,
       },
       history,
-      (delta) => {
-        appendAssistantDelta(assistantIdx, delta);
+      (event: AgentEvent) => {
+        agentEvents.push(event);
+        if (event.type === "text_delta" && event.delta) {
+          appendAssistantDelta(assistantIdx, event.delta);
+        }
+        if (event.type === "reasoning_delta" && event.delta) {
+          appendAssistantReasoning(assistantIdx, event.delta);
+        }
+        // Real-time agent step rendering
+        if (event.type === "tool_call_start" || event.type === "tool_call_end") {
+          const msg = messages.value[assistantIdx];
+          if (msg) {
+            const steps = agentEvents
+              .filter((e) => e.type === "tool_call_start" || e.type === "tool_call_end")
+              .map((e) => ({
+                key: `${e.tool_call_id || ""}-${e.type}`,
+                labelKey: e.type === "tool_call_start" ? "ai.agentSteps.callingTool" : e.is_error ? "ai.agentSteps.toolError" : "ai.agentSteps.toolDone",
+                tone: (e.type === "tool_call_start" ? "active" : e.is_error ? "danger" : "success") as AiAgentStepTone,
+                titleKey: undefined,
+                titleParams: { tool: e.tool_name || "" },
+              }));
+            msg.agentSteps = steps;
+          }
+        }
+        scrollToBottom();
       },
       sessionId,
-      (reasoningDelta) => {
-        appendAssistantReasoning(assistantIdx, reasoningDelta);
-      },
     );
-  } catch (e: any) {
-    messages.value[assistantIdx].content = `Error: ${e.message || e}`;
+  } catch (e: unknown) {
+    const message = e instanceof Error ? e.message : String(e);
+    messages.value[assistantIdx].content = `Error: ${message}`;
   } finally {
     const msg = messages.value[assistantIdx];
     if (msg) msg.isThinking = false;
     isGenerating.value = false;
-    const agentPlan = buildAiAgentPlan({
-      mode: requestedMode,
-      action: requestedAction,
-      instruction: displayText,
-      assistantContent: msg?.content || "",
-      connection: props.connection,
-    });
-    if (msg && requestedMode === "agent") msg.agentSteps = buildAiAgentStepItems(agentPlan);
-    if (agentPlan.handoffSql) emit("requestAutoExecuteSql", agentPlan.handoffSql);
+    // Render agent tool call steps from agent events
+    if (msg && agentEvents.length > 0) {
+      msg.agentSteps = agentEvents
+        .filter((e) => e.type === "tool_call_start" || e.type === "tool_call_end")
+        .map((e) => ({
+          key: `${e.tool_call_id || ""}-${e.type}`,
+          labelKey: e.type === "tool_call_start" ? "ai.agentSteps.callingTool" : e.is_error ? "ai.agentSteps.toolError" : "ai.agentSteps.toolDone",
+          tone: e.type === "tool_call_start" ? "active" : e.is_error ? "danger" : "success",
+          titleKey: undefined,
+          titleParams: { tool: e.tool_name || "" },
+        }));
+    }
+    // Fallback: use aiAgentPlan for backward compatibility
+    if (msg && !msg.agentSteps?.length) {
+      const agentPlan = buildAiAgentPlan({
+        mode: requestedMode,
+        action: requestedAction,
+        instruction: displayText,
+        assistantContent: msg?.content || "",
+        connection: props.connection,
+      });
+      if (msg && requestedMode === "agent") msg.agentSteps = buildAiAgentStepItems(agentPlan);
+      if (agentPlan.handoffSql) emit("requestAutoExecuteSql", agentPlan.handoffSql);
+    }
     activeAction.value = "generate";
     currentSessionId.value = "";
     persistConversation();
@@ -555,8 +597,9 @@ async function copyCode(code: string, key: string) {
     setTimeout(() => {
       if (copiedIndex.value === key) copiedIndex.value = "";
     }, 2000);
-  } catch (e: any) {
-    toast(t("grid.copyFailed", { message: e?.message || String(e) }), 5000);
+  } catch (e: unknown) {
+    const message = e instanceof Error ? e.message : String(e);
+    toast(t("grid.copyFailed", { message }), 5000);
   }
 }
 
@@ -620,6 +663,7 @@ onMounted(async () => {
 
 onUnmounted(() => {
   clearTimeout(mentionTimer);
+  cancelStream();
 });
 
 function triggerAction(action: AiAction, instruction?: string) {
@@ -781,7 +825,7 @@ const messageRenderer = computed(() => {
         <div v-if="connectionStore.connections.length" class="flex items-center gap-1 mb-1 text-xs text-foreground/80">
           <DatabaseIcon v-if="connection" :db-type="connectionIconType(connection)" class="h-3 w-3 shrink-0" />
           <Server v-else class="h-3 w-3 shrink-0" />
-          <Select :model-value="connection?.id || ''" @update:model-value="(v: any) => changeConnection(v)">
+          <Select :model-value="connection?.id || ''" @update:model-value="(v: any) => changeConnection(String(v ?? ''))">
             <SelectTrigger class="h-5 w-auto border-0 rounded-md bg-transparent dark:bg-transparent p-0 px-1 text-xs text-foreground/80 shadow-none focus:ring-0 focus-visible:ring-0 [&_svg]:size-3">
               <SelectValue :placeholder="t('editor.selectConnection')">{{ connection?.name || t("editor.selectConnection") }}</SelectValue>
             </SelectTrigger>
@@ -798,7 +842,7 @@ const messageRenderer = computed(() => {
             <Database class="h-3 w-3 shrink-0 text-foreground/40" />
             <Select
               :model-value="selectedDatabaseSelectValue"
-              @update:model-value="(v: any) => changeDatabase(v)"
+              @update:model-value="(v: any) => changeDatabase(String(v ?? ''))"
               @update:open="
                 (open: boolean) => {
                   if (open) loadDatabases();
diff --git a/apps/desktop/src/i18n/locales/en.ts b/apps/desktop/src/i18n/locales/en.ts
index eabaf4aab0..3cd3b36809 100644
--- a/apps/desktop/src/i18n/locales/en.ts
+++ b/apps/desktop/src/i18n/locales/en.ts
@@ -962,6 +962,9 @@
       autoExecute: "Ready to run",
       notRequested: "Run not requested",
       skipped: "Not run",
+      callingTool: "Calling tool...",
+      toolDone: "Tool completed",
+      toolError: "Tool error",
     },
     agentStepTitles: {
       riskCheck: "Risk check: {action} · {category} · {environment} · {reasons}",
diff --git a/apps/desktop/src/i18n/locales/zh-CN.ts b/apps/desktop/src/i18n/locales/zh-CN.ts
index 0e68d9847b..4803e162ba 100644
--- a/apps/desktop/src/i18n/locales/zh-CN.ts
+++ b/apps/desktop/src/i18n/locales/zh-CN.ts
@@ -961,6 +961,9 @@
       autoExecute: "准备执行",
       notRequested: "未请求执行",
       skipped: "未执行",
+      callingTool: "正在调用工具...",
+      toolDone: "工具调用完成",
+      toolError: "工具调用出错",
     },
     agentStepTitles: {
       riskCheck: "风险检查：{action} · {category} · {environment} · {reasons}",
diff --git a/apps/desktop/src/lib/ai.ts b/apps/desktop/src/lib/ai.ts
index c6aa8e3efa..ed4ef08024 100644
--- a/apps/desktop/src/lib/ai.ts
+++ b/apps/desktop/src/lib/ai.ts
@@ -7,6 +7,8 @@ import { aiTableMentionKey, type AiTableMention } from "@/lib/aiTableMentions";
 import { aiSkillForAction } from "@/lib/aiSkills";
 import { isSchemaAware } from "@/lib/databaseCapabilities";
 
+import type { AgentEvent } from "@/lib/tauri";
+
 export type AiAction = "generate" | "explain" | "optimize" | "fix" | "convert" | "sampleData";
 export type AiAssistantMode = "ask" | "agent";
 
@@ -25,6 +27,7 @@ export interface AiSchemaTable {
 }
 
 export interface AiContext {
+  connectionId: string;
   connectionName: string;
   databaseType: DatabaseType;
   database: string;
@@ -44,7 +47,7 @@ export interface AiRequestInput {
   context: AiContext;
 }
 
-export async function runAiAction(input: AiRequestInput, history?: api.AiMessage[]): Promise<string> {
+function buildAgentRequest(input: AiRequestInput, history?: api.AiMessage[]): { messages: api.AiMessage[]; systemPrompt: string; maxTokens: number; temperature: number } {
   const isZh = isChineseLocale(currentLocale());
   const skill = aiSkillForAction(input.action);
   const systemPrompt = buildSystemPrompt(input.action, input.context, input.mode);
@@ -54,27 +57,24 @@ export async function runAiAction(input: AiRequestInput, history?: api.AiMessage
   const messages: api.AiMessage[] = [...(history || []), { role: "user", content: userPrompt }];
 
   const params = actionParams(input.action);
+  const maxTokens = input.config.enableThinking ? Math.max(params.maxTokens, 8192) : params.maxTokens;
+  return { messages, systemPrompt, maxTokens, temperature: params.temperature };
+}
+
+export async function runAiAction(input: AiRequestInput, history?: api.AiMessage[]): Promise<string> {
+  const { messages, systemPrompt, maxTokens, temperature } = buildAgentRequest(input, history);
   return api.aiComplete({
     config: input.config,
     systemPrompt,
     messages,
-    maxTokens: params.maxTokens,
-    temperature: params.temperature,
+    maxTokens,
+    temperature,
   });
 }
 
 export async function runAiStream(input: AiRequestInput, history: api.AiMessage[] | undefined, onDelta: (delta: string) => void, sessionId?: string, onReasoningDelta?: (delta: string) => void): Promise<void> {
-  const isZh = isChineseLocale(currentLocale());
-  const skill = aiSkillForAction(input.action);
-  const systemPrompt = buildSystemPrompt(input.action, input.context, input.mode);
-  const instruction = isZh ? skill.userInstruction.zh : skill.userInstruction.en;
-  const userPrompt = [`Action: ${input.action}`, instruction, "", "User request:", input.instruction.trim() || "(No extra instruction provided.)"].join("\n");
-
-  const messages: api.AiMessage[] = [...(history || []), { role: "user", content: userPrompt }];
-
+  const { messages, systemPrompt, maxTokens, temperature } = buildAgentRequest(input, history);
   const sid = sessionId || uuid();
-  const params = actionParams(input.action);
-  const maxTokens = input.config.enableThinking ? Math.max(params.maxTokens, 8192) : params.maxTokens;
 
   await api.aiStream(
     sid,
@@ -83,7 +83,7 @@ export async function runAiStream(input: AiRequestInput, history: api.AiMessage[
       systemPrompt,
       messages,
       maxTokens,
-      temperature: params.temperature,
+      temperature,
     },
     (chunk) => {
       if (!chunk.done) {
@@ -94,6 +94,26 @@ export async function runAiStream(input: AiRequestInput, history: api.AiMessage[
   );
 }
 
+export async function runAgentStream(input: AiRequestInput, history: api.AiMessage[] | undefined, onEvent: (event: AgentEvent) => void, sessionId?: string): Promise<string> {
+  const { messages, systemPrompt, maxTokens, temperature } = buildAgentRequest(input, history);
+  const sid = sessionId || uuid();
+
+  return api.aiAgentStream(
+    sid,
+    {
+      config: input.config,
+      systemPrompt,
+      messages,
+      maxTokens,
+      temperature,
+    },
+    input.context.connectionId,
+    input.context.database,
+    input.context.databaseType,
+    onEvent,
+  );
+}
+
 function actionParams(action: AiAction): { maxTokens: number; temperature: number } {
   switch (action) {
     case "explain":
@@ -158,8 +178,8 @@ function buildBasePromptLines(isZh: boolean): string[] {
     isZh ? "精确、保守，根据当前数据库方言生成 SQL。" : "Be precise, conservative, and adapt SQL to the active database dialect.",
     isZh ? "严格使用当前数据库方言；标识符引用、分页、日期函数、字符串拼接、LIMIT/TOP/OFFSET 语法必须匹配数据库类型。" : "Strictly use the active database dialect; identifier quoting, pagination, date functions, string concatenation, and LIMIT/TOP/OFFSET syntax must match the database type.",
     isZh
-      ? "对于普通数据查询，优先使用下面已加载的 Schema 上下文，不要为了重复确认已给出的结构而查询 information_schema 或系统表。"
-      : "For ordinary data queries, prefer the loaded schema context below. Do not query information_schema or system tables merely to rediscover structure already provided.",
+      ? "对于普通数据查询，优先使用下面已加载的 Schema 上下文，不要为了重复确认已给出的结构而查询 information_schema 或系统表。但当用户询问某表的字段详情、列信息时，应使用 get_columns 工具获取最权威完整的定义。"
+      : "For ordinary data queries, prefer the loaded schema context below. Do not query information_schema or system tables merely to rediscover structure already provided. However, when the user asks for detailed column/field information of a specific table, use the get_columns tool for the authoritative and complete definition.",
     isZh
       ? "例外：当用户明确询问当前有哪些表/Schema、某张表是否存在、或需要盘点数据库对象时，应生成符合当前方言的只读元数据查询（例如 SHOW TABLES、information_schema、sqlite_master 等）。"
       : "Exception: when the user explicitly asks what tables/schemas exist, whether a table exists, or asks for database object inventory, generate a read-only metadata query appropriate for the active dialect (for example SHOW TABLES, information_schema, sqlite_master).",
@@ -314,6 +334,7 @@ export async function buildAiContext(tab: QueryTab, connection: ConnectionConfig
   }
 
   return {
+    connectionId: tab.connectionId,
     connectionName: connection.name,
     databaseType: connection.db_type,
     database: tab.database,
diff --git a/apps/desktop/src/lib/api.ts b/apps/desktop/src/lib/api.ts
index f54ca034d7..59927ebbbe 100644
--- a/apps/desktop/src/lib/api.ts
+++ b/apps/desktop/src/lib/api.ts
@@ -174,6 +174,7 @@ export const buildDataCompareSyncPlan = forward("buildDataCompareSyncPlan");
 // AI
 export const aiComplete = forward("aiComplete");
 export const aiStream = forward("aiStream");
+export const aiAgentStream = forward("aiAgentStream");
 export const aiCancelStream = forward("aiCancelStream");
 export const aiTestConnection = forward("aiTestConnection");
 export const aiListModels = forward("aiListModels");
@@ -353,4 +354,5 @@ export type {
   TableExportProgress,
   TableExportStatus,
   TableExportRequest,
+  AgentEvent,
 } from "./tauri";
diff --git a/apps/desktop/src/lib/http.ts b/apps/desktop/src/lib/http.ts
index a0b0fbcc1c..8d73638150 100644
--- a/apps/desktop/src/lib/http.ts
+++ b/apps/desktop/src/lib/http.ts
@@ -757,6 +757,58 @@ export async function aiListModels(config: AiConfig): Promise<AiModelInfo[]> {
   return post("/api/ai/models", { config });
 }
 
+export type { AgentEvent } from "./tauri";
+
+function isAgentEvent(v: unknown): v is import("./tauri").AgentEvent {
+  return typeof v === "object" && v !== null && "type" in v && typeof (v as Record<string, unknown>).type === "string";
+}
+
+export async function aiAgentStream(sessionId: string, request: AiCompletionRequest, connectionId: string, database: string, dbType: string, onEvent: (event: import("./tauri").AgentEvent) => void, signal?: AbortSignal): Promise<string> {
+  const res = await fetch("/api/ai/agent-stream", {
+    method: "POST",
+    headers: { "Content-Type": "application/json" },
+    body: JSON.stringify({ session_id: sessionId, request, connection_id: connectionId, database, db_type: dbType }),
+    signal,
+  });
+  if (!res.ok) throw new Error(await res.text());
+
+  const reader = res.body!.getReader();
+  const decoder = new TextDecoder();
+  let buffer = "";
+  let result = "";
+
+  while (true) {
+    const { done, value } = await reader.read();
+    if (done) break;
+    buffer += decoder.decode(value, { stream: true });
+
+    const lines = buffer.split("\n");
+    buffer = lines.pop() || "";
+
+    for (const line of lines) {
+      if (line.startsWith("data:")) {
+        const data = line.slice(5).trim();
+        if (data && data !== "[DONE]") {
+          try {
+            const parsed = JSON.parse(data);
+            if (!isAgentEvent(parsed)) {
+              console.warn("[aiAgentStream] Skipping invalid agent event:", data);
+              continue;
+            }
+            onEvent(parsed);
+            if (parsed.type === "agent_end" || parsed.type === "error") {
+              result = data;
+            }
+          } catch {
+            // skip malformed JSON
+          }
+        }
+      }
+    }
+  }
+  return result;
+}
+
 export async function saveAiConfig(config: AiConfig): Promise<void> {
   return post("/api/ai/config", { config });
 }
diff --git a/apps/desktop/src/lib/tauri.ts b/apps/desktop/src/lib/tauri.ts
index e62583eac0..5b3269b326 100644
--- a/apps/desktop/src/lib/tauri.ts
+++ b/apps/desktop/src/lib/tauri.ts
@@ -274,6 +274,31 @@ export async function aiStream(sessionId: string, request: AiCompletionRequest,
   }
 }
 
+export type AgentEvent =
+  | { type: "turn_start"; turn: number }
+  | { type: "text_delta"; delta: string }
+  | { type: "reasoning_delta"; delta: string }
+  | { type: "tool_call_start"; tool_call_id: string; tool_name: string; args: Record<string, unknown> }
+  | { type: "tool_call_end"; tool_call_id: string; tool_name: string; result: unknown; is_error: boolean }
+  | { type: "turn_end"; turn: number }
+  | { type: "agent_end"; total_tokens?: number }
+  | { type: "error"; message: string };
+
+export async function aiAgentStream(sessionId: string, request: AiCompletionRequest, connectionId: string, database: string, dbType: string, onEvent: (event: AgentEvent) => void): Promise<string> {
+  const unlisten: UnlistenFn = await listen<AgentEvent>("ai-agent-event", (event) => {
+    onEvent(event.payload);
+    if (event.payload.type === "agent_end" || event.payload.type === "error") {
+      unlisten();
+    }
+  });
+  try {
+    return await invoke("ai_agent_stream", { sessionId, request, connectionId, database, dbType });
+  } catch (e) {
+    unlisten();
+    throw e;
+  }
+}
+
 export async function saveAiConfig(config: AiConfig): Promise<void> {
   return invoke("save_ai_config", { config });
 }
diff --git a/crates/dbx-core/src/agent_events.rs b/crates/dbx-core/src/agent_events.rs
new file mode 100644
index 0000000000..91ea8526c6
--- /dev/null
+++ b/crates/dbx-core/src/agent_events.rs
@@ -0,0 +1,79 @@
+use serde::{Deserialize, Serialize};
+
+/// Events emitted by the agent loop, streamed to the frontend via SSE.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(tag = "type", rename_all = "snake_case")]
+pub enum AgentEvent {
+    /// A turn in the agent loop has started.
+    TurnStart { turn: u32 },
+    /// Text delta from the LLM response.
+    TextDelta { delta: String },
+    /// Reasoning/thinking delta (for models that support it).
+    ReasoningDelta { delta: String },
+    /// The LLM wants to call a tool.
+    ToolCallStart { tool_call_id: String, tool_name: String, args: serde_json::Value },
+    /// A tool execution has completed.
+    ToolCallEnd { tool_call_id: String, tool_name: String, result: serde_json::Value, is_error: bool },
+    /// A turn in the agent loop has ended.
+    TurnEnd { turn: u32 },
+    /// The agent loop has finished.
+    AgentEnd { total_tokens: Option<u64> },
+    /// An error occurred during the agent loop.
+    Error { message: String },
+}
+
+/// A unified tool call extracted from any provider's response.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ToolCall {
+    pub id: String,
+    pub name: String,
+    pub arguments: serde_json::Value,
+}
+
+/// Result of executing a tool.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ToolResult {
+    pub tool_call_id: String,
+    pub tool_name: String,
+    pub content: String,
+    pub is_error: bool,
+}
+
+/// Definition of a tool available to the agent.
+#[derive(Debug, Clone)]
+pub struct ToolDefinition {
+    pub name: &'static str,
+    pub description: &'static str,
+    pub parameters: serde_json::Value,
+    pub read_only: bool,
+}
+
+impl ToolDefinition {
+    /// Convert to OpenAI-compatible tool JSON for the API request.
+    pub fn to_openai_tool(&self) -> serde_json::Value {
+        serde_json::json!({
+            "type": "function",
+            "function": {
+                "name": self.name,
+                "description": self.description,
+                "parameters": self.parameters
+            }
+        })
+    }
+    /// Convert to Anthropic-compatible tool JSON.
+    pub fn to_anthropic_tool(&self) -> serde_json::Value {
+        serde_json::json!({
+            "name": self.name,
+            "description": self.description,
+            "input_schema": self.parameters
+        })
+    }
+    /// Convert to Gemini-compatible function declaration.
+    pub fn to_gemini_tool(&self) -> serde_json::Value {
+        serde_json::json!({
+            "name": self.name,
+            "description": self.description,
+            "parameters": self.parameters
+        })
+    }
+}
diff --git a/crates/dbx-core/src/agent_loop.rs b/crates/dbx-core/src/agent_loop.rs
new file mode 100644
index 0000000000..f3a1e0776b
--- /dev/null
+++ b/crates/dbx-core/src/agent_loop.rs
@@ -0,0 +1,280 @@
+use std::sync::Arc;
+
+use futures::FutureExt;
+use serde_json::json;
+use tokio::sync::Notify;
+
+use crate::agent_events::{AgentEvent, ToolCall, ToolDefinition};
+use crate::agent_tools;
+use crate::ai::{self, AiCompletionRequest, AiConfig, AiMessage, AiProvider, AiStreamChunk};
+use crate::connection::AppState;
+use crate::models::connection::DatabaseType;
+use tokio::sync::Mutex;
+
+/// Maximum number of agent loop turns to prevent infinite loops.
+const MAX_AGENT_TURNS: u32 = 10;
+
+/// Context for an agent loop run.
+pub struct AgentLoopContext {
+    pub state: Arc<AppState>,
+    pub connection_id: String,
+    pub database: String,
+    pub db_type: DatabaseType,
+}
+
+/// Check if the provider supports function calling / tool use.
+/// Returns false for providers that are known to lack reliable tool support.
+fn provider_supports_function_calling(config: &AiConfig) -> bool {
+    match config.provider {
+        // Ollama function calling support varies by model/version; conservative default is false.
+        // Users with capable models can override via openai-compatible with an Ollama endpoint.
+        AiProvider::Ollama => false,
+        _ => true,
+    }
+}
+
+/// Run the agent loop: call LLM with tools, execute tool calls, feed results back, repeat.
+///
+/// The `on_event` callback receives streaming events for the frontend.
+/// Returns the final accumulated assistant text.
+///
+/// If the provider does not support function calling (e.g., Ollama), automatically
+/// degrades to a text-only completion with schema context injected into the system prompt.
+pub async fn run_agent_loop(
+    config: &AiConfig,
+    system_prompt: &str,
+    messages: &[AiMessage],
+    agent_ctx: &AgentLoopContext,
+    on_event: impl Fn(AgentEvent) + Send + Sync + Clone + 'static,
+    cancelled: &Notify,
+    max_tokens: Option<u32>,
+    temperature: Option<f32>,
+) -> Result<String, String> {
+    // Auto-degrade: providers without function calling fall back to text-only completion.
+    if !provider_supports_function_calling(config) {
+        return run_agent_loop_text_only(
+            config,
+            system_prompt,
+            messages,
+            agent_ctx,
+            on_event,
+            cancelled,
+            max_tokens,
+            temperature,
+        )
+        .await;
+    }
+    let tools = agent_tools::read_only_tools();
+    let mut conversation_messages: Vec<AiMessage> = messages.to_vec();
+    let mut final_text = String::new();
+
+    for turn in 0..MAX_AGENT_TURNS {
+        // Check for cancellation before each turn
+        if cancelled.notified().now_or_never().is_some() {
+            on_event(AgentEvent::Error { message: "Agent loop cancelled".to_string() });
+            break;
+        }
+
+        on_event(AgentEvent::TurnStart { turn });
+
+        // Build the LLM request with tools
+        let request =
+            build_tool_request(config, system_prompt, &conversation_messages, &tools, max_tokens, temperature);
+
+        // Stream the LLM response, collecting text and tool_calls
+        let accumulated_text = Arc::new(Mutex::new(String::new()));
+        let session_id = format!("agent-turn-{turn}");
+
+        let acc = accumulated_text.clone();
+        let on_event2 = on_event.clone();
+        let on_chunk = move |chunk: AiStreamChunk| {
+            if !chunk.delta.is_empty() {
+                if let Ok(mut text) = acc.try_lock() {
+                    text.push_str(&chunk.delta);
+                }
+                on_event2(AgentEvent::TextDelta { delta: chunk.delta.clone() });
+            }
+            if let Some(ref reasoning) = chunk.reasoning_delta {
+                on_event2(AgentEvent::ReasoningDelta { delta: reasoning.clone() });
+            }
+        };
+
+        // Call the LLM with tool support
+        let collected_tool_calls = stream_with_tools(config, &request, &session_id, cancelled, on_chunk).await?;
+
+        on_event(AgentEvent::TurnEnd { turn });
+
+        let accumulated_text = accumulated_text.lock().await.clone();
+
+        // Add assistant message to conversation (including tool_use blocks)
+        conversation_messages.push(AiMessage {
+            role: "assistant".to_string(),
+            content: accumulated_text.clone(),
+            tool_call_id: None,
+            tool_calls: collected_tool_calls
+                .iter()
+                .map(|tc| ai::ToolCallRef { id: tc.id.clone(), name: tc.name.clone(), arguments: tc.arguments.clone() })
+                .collect(),
+        });
+
+        if collected_tool_calls.is_empty() {
+            // No tool calls -- we're done
+            final_text = accumulated_text;
+            break;
+        }
+
+        // Execute each tool call
+        for tc in &collected_tool_calls {
+            on_event(AgentEvent::ToolCallStart {
+                tool_call_id: tc.id.clone(),
+                tool_name: tc.name.clone(),
+                args: tc.arguments.clone(),
+            });
+
+            let result = agent_tools::execute_tool(
+                tc,
+                &agent_ctx.state,
+                &agent_ctx.connection_id,
+                &agent_ctx.database,
+                &agent_ctx.db_type,
+            )
+            .await;
+
+            on_event(AgentEvent::ToolCallEnd {
+                tool_call_id: tc.id.clone(),
+                tool_name: tc.name.clone(),
+                result: json!({ "content": result.content }),
+                is_error: result.is_error,
+            });
+
+            // Add tool result to conversation for the next LLM call
+            // Uses "tool" role per OpenAI convention; provider-specific conversion
+            // happens in ai::stream_with_tools when building provider API requests.
+            conversation_messages.push(AiMessage {
+                role: "tool".to_string(),
+                content: result.content.clone(),
+                tool_call_id: Some(tc.id.clone()),
+                tool_calls: Vec::new(),
+            });
+        }
+
+        final_text = accumulated_text;
+    }
+
+    on_event(AgentEvent::AgentEnd { total_tokens: None });
+    Ok(final_text)
+}
+
+/// Build an LLM request that includes tool definitions.
+fn build_tool_request(
+    config: &AiConfig,
+    system_prompt: &str,
+    messages: &[AiMessage],
+    _tools: &[ToolDefinition], // Tools are injected in ai::stream_with_tools, not via AiCompletionRequest.
+    max_tokens: Option<u32>,
+    temperature: Option<f32>,
+) -> AiCompletionRequest {
+    // Note: tools are passed via the body, not via AiCompletionRequest.
+    // The actual injection happens in stream_with_tools.
+    AiCompletionRequest {
+        config: config.clone(),
+        system_prompt: system_prompt.to_string(),
+        messages: messages.to_vec(),
+        max_tokens: max_tokens.or(Some(4096)),
+        temperature: temperature.or(Some(0.2)),
+    }
+}
+
+/// Stream an LLM response with tool support, parsing tool_calls from SSE deltas.
+///
+/// True streaming: text, reasoning, and tool call arguments are all emitted
+/// incrementally as they arrive from the provider.
+async fn stream_with_tools(
+    config: &AiConfig,
+    request: &AiCompletionRequest,
+    session_id: &str,
+    cancelled: &Notify,
+    on_chunk: impl Fn(AiStreamChunk) + Send + Sync + 'static,
+) -> Result<Vec<ToolCall>, String> {
+    let tools = agent_tools::read_only_tools();
+
+    // Return early if the user cancelled before the LLM call started.
+    if cancelled.notified().now_or_never().is_some() {
+        return Err("Agent loop cancelled".to_string());
+    }
+
+    ai::stream_with_tools(config, request, session_id, &tools, cancelled, on_chunk).await
+}
+
+/// Text-only fallback for providers that don't support function calling.
+///
+/// Injects database schema context into the system prompt so the LLM can still
+/// give informed answers, then performs a single non-streaming completion.
+async fn run_agent_loop_text_only(
+    config: &AiConfig,
+    system_prompt: &str,
+    messages: &[AiMessage],
+    agent_ctx: &AgentLoopContext,
+    on_event: impl Fn(AgentEvent) + Send + Sync + 'static,
+    _cancelled: &Notify,
+    max_tokens: Option<u32>,
+    temperature: Option<f32>,
+) -> Result<String, String> {
+    // Build a schema-enriched system prompt so the LLM can answer schema questions
+    // even without tool access.
+    let enriched_prompt = build_schema_prompt(agent_ctx, system_prompt).await;
+
+    let request = AiCompletionRequest {
+        config: config.clone(),
+        system_prompt: enriched_prompt,
+        messages: messages.to_vec(),
+        max_tokens: max_tokens.or(Some(4096)),
+        temperature: temperature.or(Some(0.2)),
+    };
+
+    // Use a non-streaming completion as the simplest fallback.
+    let result = ai::complete(&request).await?;
+
+    on_event(AgentEvent::TextDelta { delta: result.clone() });
+    on_event(AgentEvent::AgentEnd { total_tokens: None });
+    Ok(result)
+}
+
+/// Build a system prompt enriched with database schema information
+/// for text-only mode where the LLM cannot use tools.
+async fn build_schema_prompt(agent_ctx: &AgentLoopContext, system_prompt: &str) -> String {
+    let mut enriched = system_prompt.to_string();
+
+    // Fetch real schema data using the same core functions the tools would use
+    let tables_result = crate::schema::list_tables_core(
+        &agent_ctx.state,
+        &agent_ctx.connection_id,
+        &agent_ctx.database,
+        "",
+        None,
+        Some(50), // smaller limit for prompt injection
+    )
+    .await;
+
+    match tables_result {
+        Ok(tables) if !tables.is_empty() => {
+            enriched.push_str("\n\n## Database Schema (for context — no tools available)\n");
+            enriched.push_str(&format!("Database: {}\n", agent_ctx.database));
+            enriched.push_str("Tables:\n");
+            for t in &tables {
+                enriched.push_str(&format!("  - {} ({})", t.name, t.table_type));
+                if let Some(ref comment) = t.comment {
+                    if !comment.trim().is_empty() {
+                        enriched.push_str(&format!(" — {}", comment.trim()));
+                    }
+                }
+                enriched.push('\n');
+            }
+        }
+        _ => {
+            enriched.push_str("\n\n(Note: Unable to load database schema for this request.)\n");
+        }
+    }
+
+    enriched
+}
diff --git a/crates/dbx-core/src/agent_tools.rs b/crates/dbx-core/src/agent_tools.rs
new file mode 100644
index 0000000000..c8d8f5b4bc
--- /dev/null
+++ b/crates/dbx-core/src/agent_tools.rs
@@ -0,0 +1,204 @@
+use std::sync::Arc;
+
+use serde_json::json;
+
+use crate::agent_events::{ToolCall, ToolDefinition, ToolResult};
+use crate::connection::AppState;
+use crate::models::connection::DatabaseType;
+
+/// Maximum number of tables returned by list_tables tool.
+const LIST_TABLES_LIMIT: usize = 200;
+
+/// Get all available tool definitions for Phase 1 (read-only tools).
+pub fn read_only_tools() -> Vec<ToolDefinition> {
+    vec![list_tables_tool(), get_columns_tool()]
+}
+
+/// list_tables tool definition.
+fn list_tables_tool() -> ToolDefinition {
+    ToolDefinition {
+        name: "list_tables",
+        description: "List all tables and views in the current database. Returns table names, types, and comments.",
+        parameters: json!({
+            "type": "object",
+            "properties": {
+                "schema": {
+                    "type": "string",
+                    "description": "Schema name to list tables from (optional, defaults to current database)"
+                }
+            },
+            "required": []
+        }),
+        read_only: true,
+    }
+}
+
+/// get_columns tool definition.
+fn get_columns_tool() -> ToolDefinition {
+    ToolDefinition {
+        name: "get_columns",
+        description:
+            "Get column definitions for a table: names, types, primary keys, nullable, defaults, and comments. \
+             Use this when the user asks about table structure, column details, or field information — \
+             even if some schema context was provided, this tool returns the authoritative and complete column list.",
+        parameters: json!({
+            "type": "object",
+            "properties": {
+                "table": {
+                    "type": "string",
+                    "description": "Table name to get columns for"
+                },
+                "schema": {
+                    "type": "string",
+                    "description": "Schema name (optional, defaults to current database)"
+                }
+            },
+            "required": ["table"]
+        }),
+        read_only: true,
+    }
+}
+
+/// Execute a tool call and return the result.
+pub async fn execute_tool(
+    tool_call: &ToolCall,
+    state: &Arc<AppState>,
+    connection_id: &str,
+    database: &str,
+    db_type: &DatabaseType,
+) -> ToolResult {
+    let result = match tool_call.name.as_str() {
+        "list_tables" => execute_list_tables(tool_call, state, connection_id, database, db_type).await,
+        "get_columns" => execute_get_columns(tool_call, state, connection_id, database, db_type).await,
+        _ => Err(format!("Unknown tool: {}", tool_call.name)),
+    };
+
+    match result {
+        Ok(content) => ToolResult {
+            tool_call_id: tool_call.id.clone(),
+            tool_name: tool_call.name.clone(),
+            content,
+            is_error: false,
+        },
+        Err(err) => ToolResult {
+            tool_call_id: tool_call.id.clone(),
+            tool_name: tool_call.name.clone(),
+            content: format!("Error: {err}"),
+            is_error: true,
+        },
+    }
+}
+
+async fn execute_list_tables(
+    tool_call: &ToolCall,
+    state: &Arc<AppState>,
+    connection_id: &str,
+    database: &str,
+    _db_type: &DatabaseType,
+) -> Result<String, String> {
+    let schema = tool_call.arguments.get("schema").and_then(|v| v.as_str()).unwrap_or("").to_string();
+
+    // list_tables_core already limits via the Some(LIST_TABLES_LIMIT) argument.
+    let tables =
+        crate::schema::list_tables_core(state, connection_id, database, &schema, None, Some(LIST_TABLES_LIMIT))
+            .await
+            .map_err(|e| format!("Failed to list tables: {e}"))?;
+
+    let total = tables.len();
+
+    let mut lines = Vec::new();
+    for table in &tables {
+        let mut line = format!("- {} ({})", table.name, table.table_type);
+        if let Some(comment) = &table.comment {
+            let trimmed = comment.trim();
+            if !trimmed.is_empty() {
+                line.push_str(&format!(" -- {}", trimmed));
+            }
+        }
+        lines.push(line);
+    }
+
+    if total > LIST_TABLES_LIMIT {
+        lines.push(format!("... (showing {LIST_TABLES_LIMIT} of {total} tables)"));
+    }
+
+    if lines.is_empty() {
+        return Ok("No tables found in this database/schema.".to_string());
+    }
+
+    Ok(lines.join("\n"))
+}
+
+async fn execute_get_columns(
+    tool_call: &ToolCall,
+    state: &Arc<AppState>,
+    connection_id: &str,
+    database: &str,
+    _db_type: &DatabaseType,
+) -> Result<String, String> {
+    let table = tool_call
+        .arguments
+        .get("table")
+        .and_then(|v| v.as_str())
+        .ok_or("Missing required parameter: table")?
+        .trim()
+        .to_string();
+
+    if table.is_empty() {
+        return Err("Table name cannot be empty".to_string());
+    }
+    if table.len() > 256 {
+        return Err(format!("Table name too long: {} characters (max 256)", table.len()));
+    }
+    // Reject names with characters that are unlikely to be valid identifiers
+    if table.contains(';') || table.contains('\'') || table.contains('"') || table.contains('\\') {
+        return Err(format!("Table name contains invalid characters: '{}'", table));
+    }
+
+    let schema = tool_call.arguments.get("schema").and_then(|v| v.as_str()).unwrap_or("").to_string();
+
+    let columns = crate::schema::get_columns_core(state, connection_id, database, &schema, &table)
+        .await
+        .map_err(|e| format!("Failed to get columns for {table}: {e}"))?;
+
+    if columns.is_empty() {
+        return Ok(format!("No columns found for table '{table}'."));
+    }
+
+    let mut lines = Vec::new();
+    lines.push(format!("Columns of {table}:"));
+    for col in &columns {
+        let mut flags: Vec<String> = Vec::new();
+        if col.is_primary_key {
+            flags.push("PK".to_string());
+        }
+        if col.is_nullable {
+            flags.push("nullable".to_string());
+        } else {
+            flags.push("NOT NULL".to_string());
+        }
+        if let Some(default) = &col.column_default {
+            if !default.is_empty() {
+                flags.push(format!("default {default}"));
+            }
+        }
+        if let Some(extra) = &col.extra {
+            if !extra.is_empty() {
+                flags.push(extra.clone());
+            }
+        }
+
+        let flags_str = if flags.is_empty() { String::new() } else { format!(" ({})", flags.join(", ")) };
+
+        let comment_str = col
+            .comment
+            .as_ref()
+            .filter(|c| !c.trim().is_empty())
+            .map(|c| format!(" -- {}", c.trim()))
+            .unwrap_or_default();
+
+        lines.push(format!("  - {}: {}{}{}", col.name, col.data_type, flags_str, comment_str));
+    }
+
+    Ok(lines.join("\n"))
+}
diff --git a/crates/dbx-core/src/ai.rs b/crates/dbx-core/src/ai.rs
index 848b77e4db..7bb25c615b 100644
--- a/crates/dbx-core/src/ai.rs
+++ b/crates/dbx-core/src/ai.rs
@@ -99,6 +99,26 @@ fn default_enable_thinking() -> bool {
 pub struct AiMessage {
     pub role: String,
     pub content: String,
+    /// Tool call ID for tool results (role="tool"). Used to associate
+    /// a tool result with its originating tool call in multi-turn loops.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub tool_call_id: Option<String>,
+    /// Tool calls made by the assistant (role="assistant"). Used to
+    /// reconstruct tool_use content blocks for providers like Anthropic
+    /// that require them in the conversation history.
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub tool_calls: Vec<ToolCallRef>,
+}
+
+/// A lightweight reference to a tool call within an assistant message.
+/// Stores the id, name, and arguments needed to reconstruct provider-specific
+/// tool_use content blocks (e.g. Anthropic's `{"type":"tool_use", ...}`).
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ToolCallRef {
+    pub id: String,
+    pub name: String,
+    pub arguments: serde_json::Value,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -313,7 +333,7 @@ pub fn supports_temperature(config: &AiConfig) -> bool {
     !(is_openai_api_config(config) && is_openai_reasoning_model(&config.model))
 }
 
-fn add_temperature_if_supported(body: &mut serde_json::Value, request: &AiCompletionRequest) {
+pub fn add_temperature_if_supported(body: &mut serde_json::Value, request: &AiCompletionRequest) {
     if supports_temperature(&request.config) {
         body["temperature"] = json!(request.temperature.unwrap_or(0.2));
     }
@@ -388,7 +408,7 @@ fn validate_model_list_config(config: &AiConfig) -> Result<(), String> {
     resolve_model_list_endpoint(config).map(|_| ())
 }
 
-fn maybe_bearer_headers(config: &AiConfig) -> Result<HeaderMap, String> {
+pub fn maybe_bearer_headers(config: &AiConfig) -> Result<HeaderMap, String> {
     let mut headers = HeaderMap::new();
     headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
     if !config.api_key.trim().is_empty() {
@@ -400,7 +420,7 @@ fn maybe_bearer_headers(config: &AiConfig) -> Result<HeaderMap, String> {
     Ok(headers)
 }
 
-fn claude_headers(config: &AiConfig) -> Result<HeaderMap, String> {
+pub fn claude_headers(config: &AiConfig) -> Result<HeaderMap, String> {
     let mut headers = HeaderMap::new();
     headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
     match config.auth_method {
@@ -680,7 +700,12 @@ pub async fn test_connection_core(config: &AiConfig) -> Result<String, String> {
     let request = AiCompletionRequest {
         config: config.clone(),
         system_prompt: String::new(),
-        messages: vec![AiMessage { role: "user".into(), content: "hi".into() }],
+        messages: vec![AiMessage {
+            role: "user".into(),
+            content: "hi".into(),
+            tool_call_id: None,
+            tool_calls: Vec::new(),
+        }],
         max_tokens: Some(16),
         temperature: Some(0.0),
     };
@@ -1095,6 +1120,570 @@ async fn stream_gemini(
     Ok(())
 }
 
+// ---------------------------------------------------------------------------
+// Streaming with tools (agent loop)
+// ---------------------------------------------------------------------------
+
+/// Events emitted by provider-specific streaming-with-tools functions.
+/// The public `stream_with_tools` entry point uses these to accumulate
+/// tool calls and forward text/reasoning chunks to the caller.
+pub enum StreamToolEvent {
+    /// A text or reasoning delta for the frontend.
+    Chunk(AiStreamChunk),
+    /// A tool_use / function_call block has started.
+    ToolCallStart { index: u32, id: String, name: String },
+    /// An argument fragment for an in-progress tool call.
+    ToolCallDelta { index: u32, fragment: String },
+    /// A tool_use / function_call block has ended.
+    ToolCallComplete { index: u32 },
+}
+
+/// Partially accumulated tool call during streaming.
+#[derive(Debug)]
+struct PartialToolCall {
+    id: String,
+    name: String,
+    arguments: String,
+}
+
+/// Accumulates streaming tool-call events into complete `ToolCall` objects.
+#[derive(Debug)]
+pub struct StreamingToolCallAccumulator {
+    calls: std::collections::HashMap<u32, PartialToolCall>,
+    ordered_indices: Vec<u32>,
+}
+
+impl StreamingToolCallAccumulator {
+    pub fn new() -> Self {
+        Self { calls: std::collections::HashMap::new(), ordered_indices: Vec::new() }
+    }
+
+    pub fn process(&mut self, event: StreamToolEvent, on_chunk: &impl Fn(AiStreamChunk)) {
+        match event {
+            StreamToolEvent::Chunk(chunk) => on_chunk(chunk),
+            StreamToolEvent::ToolCallStart { index, id, name } => {
+                self.calls.insert(index, PartialToolCall { id, name, arguments: String::new() });
+                if !self.ordered_indices.contains(&index) {
+                    self.ordered_indices.push(index);
+                }
+            }
+            StreamToolEvent::ToolCallDelta { index, fragment } => {
+                if let Some(tc) = self.calls.get_mut(&index) {
+                    tc.arguments.push_str(&fragment);
+                }
+            }
+            StreamToolEvent::ToolCallComplete { index: _ } => {
+                // Nothing extra to do — the call is already accumulated.
+            }
+        }
+    }
+
+    pub fn finalize(self) -> Vec<crate::agent_events::ToolCall> {
+        let mut result = Vec::new();
+        for idx in &self.ordered_indices {
+            if let Some(tc) = self.calls.get(idx) {
+                let args: serde_json::Value = serde_json::from_str(&tc.arguments).unwrap_or(json!({}));
+                result.push(crate::agent_events::ToolCall {
+                    id: tc.id.clone(),
+                    name: tc.name.clone(),
+                    arguments: args,
+                });
+            }
+        }
+        result
+    }
+}
+
+/// Streaming Claude call with tool support.
+/// Returns a stream of `StreamToolEvent` via the `on_event` callback.
+async fn stream_claude_with_tools(
+    client: &reqwest::Client,
+    session_id: &str,
+    request: &AiCompletionRequest,
+    tools: &[crate::agent_events::ToolDefinition],
+    cancelled: &Notify,
+    on_event: &impl Fn(StreamToolEvent),
+) -> Result<(), String> {
+    let mut messages: Vec<serde_json::Value> = Vec::new();
+    let mut pending_tool_results: Vec<serde_json::Value> = Vec::new();
+    for m in &request.messages {
+        if m.role == "tool" {
+            // Collect consecutive tool results; flush as a single user message.
+            pending_tool_results.push(json!({
+                "type": "tool_result",
+                "tool_use_id": m.tool_call_id.as_deref().unwrap_or_default(),
+                "content": m.content
+            }));
+        } else {
+            // Flush any pending tool results before emitting a non-tool message.
+            if !pending_tool_results.is_empty() {
+                messages.push(json!({
+                    "role": "user",
+                    "content": pending_tool_results.drain(..).collect::<Vec<_>>()
+                }));
+            }
+            if m.role == "assistant" && !m.tool_calls.is_empty() {
+                let mut content_blocks: Vec<serde_json::Value> = Vec::new();
+                if !m.content.is_empty() {
+                    content_blocks.push(json!({ "type": "text", "text": m.content }));
+                }
+                for tc in &m.tool_calls {
+                    content_blocks
+                        .push(json!({ "type": "tool_use", "id": tc.id, "name": tc.name, "input": tc.arguments }));
+                }
+                messages.push(json!({ "role": "assistant", "content": content_blocks }));
+            } else {
+                messages.push(json!({ "role": m.role, "content": m.content }));
+            }
+        }
+    }
+    // Flush any remaining tool results at the end of the message list.
+    if !pending_tool_results.is_empty() {
+        messages.push(json!({
+            "role": "user",
+            "content": pending_tool_results.drain(..).collect::<Vec<_>>()
+        }));
+    }
+
+    let tool_json: Vec<serde_json::Value> = tools.iter().map(|t| t.to_anthropic_tool()).collect();
+
+    let mut body = json!({
+        "model": request.config.model,
+        "max_tokens": request.max_tokens.unwrap_or(4096),
+        "system": request.system_prompt,
+        "messages": messages,
+        "tools": tool_json,
+        "stream": true,
+    });
+    add_temperature_if_supported(&mut body, request);
+
+    let res = client
+        .post(resolve_endpoint(&request.config))
+        .headers(claude_headers(&request.config)?)
+        .json(&body)
+        .send()
+        .await
+        .map_err(|e| format!("Claude request failed: {e}"))?;
+
+    if !res.status().is_success() {
+        let data: serde_json::Value = res.json().await.map_err(|e| e.to_string())?;
+        return Err(extract_error(&data).unwrap_or_else(|| "Claude API error".to_string()));
+    }
+
+    let mut byte_stream = res.bytes_stream();
+    let mut buf = String::new();
+    // Track the current content block index and type for tool_use blocks
+    let mut current_block_index: Option<u32> = None;
+    let mut current_block_type: Option<String> = None;
+
+    loop {
+        tokio::select! {
+            chunk = byte_stream.next() => {
+                let Some(chunk) = chunk else { break };
+                let chunk = chunk.map_err(|e| e.to_string())?;
+                buf.push_str(&String::from_utf8_lossy(&chunk));
+
+                let mut finished = false;
+                while let Some(pos) = buf.find('\n') {
+                    let line = buf[..pos].to_string();
+                    buf = buf[pos + 1..].to_string();
+
+                    let Some(data) = stream_data_payload(&line) else { continue };
+                    if data == "[DONE]" {
+                        finished = true;
+                        break;
+                    }
+
+                    if let Ok(event) = serde_json::from_str::<serde_json::Value>(data) {
+                        let event_type = event["type"].as_str().unwrap_or("");
+
+                        match event_type {
+                            "content_block_start" => {
+                                let idx = event["index"].as_u64().unwrap_or(0) as u32;
+                                let block_type = event["content_block"]["type"].as_str().unwrap_or("");
+                                current_block_index = Some(idx);
+                                current_block_type = Some(block_type.to_string());
+
+                                if block_type == "tool_use" {
+                                    let id = event["content_block"]["id"].as_str().unwrap_or_default().to_string();
+                                    let name = event["content_block"]["name"].as_str().unwrap_or_default().to_string();
+                                    on_event(StreamToolEvent::ToolCallStart { index: idx, id, name });
+                                }
+                            }
+                            "content_block_delta" => {
+                                let idx = event["index"].as_u64().unwrap_or(0) as u32;
+                                let delta_type = event["delta"]["type"].as_str().unwrap_or("");
+
+                                match delta_type {
+                                    "text_delta" => {
+                                        if let Some(text) = event["delta"]["text"].as_str() {
+                                            on_event(StreamToolEvent::Chunk(AiStreamChunk {
+                                                session_id: session_id.to_string(),
+                                                delta: text.to_string(),
+                                                reasoning_delta: None,
+                                                done: false,
+                                            }));
+                                        }
+                                    }
+                                    "thinking_delta" => {
+                                        if let Some(thinking) = event["delta"]["thinking"].as_str() {
+                                            on_event(StreamToolEvent::Chunk(AiStreamChunk {
+                                                session_id: session_id.to_string(),
+                                                delta: String::new(),
+                                                reasoning_delta: Some(thinking.to_string()),
+                                                done: false,
+                                            }));
+                                        }
+                                    }
+                                    "input_json_delta" => {
+                                        if let Some(fragment) = event["delta"]["partial_json"].as_str() {
+                                            on_event(StreamToolEvent::ToolCallDelta {
+                                                index: idx,
+                                                fragment: fragment.to_string(),
+                                            });
+                                        }
+                                    }
+                                    _ => {}
+                                }
+                            }
+                            "content_block_stop" => {
+                                if let Some(idx) = current_block_index.take() {
+                                    if current_block_type.as_deref() == Some("tool_use") {
+                                        on_event(StreamToolEvent::ToolCallComplete { index: idx });
+                                    }
+                                }
+                                current_block_type = None;
+                            }
+                            _ => {}
+                        }
+                    }
+                }
+
+                if finished { break; }
+            }
+            _ = cancelled.notified() => { break; }
+        }
+    }
+
+    Ok(())
+}
+
+/// Streaming OpenAI-compatible call with tool support.
+async fn stream_openai_with_tools(
+    client: &reqwest::Client,
+    session_id: &str,
+    request: &AiCompletionRequest,
+    tools: &[crate::agent_events::ToolDefinition],
+    cancelled: &Notify,
+    on_event: &impl Fn(StreamToolEvent),
+) -> Result<(), String> {
+    let headers = maybe_bearer_headers(&request.config)?;
+
+    let mut messages = vec![json!({ "role": "system", "content": request.system_prompt })];
+    messages.extend(request.messages.iter().map(|m| {
+        let mut msg = json!({ "role": m.role, "content": m.content });
+        if m.role == "tool" {
+            if let Some(ref tc_id) = m.tool_call_id {
+                msg["tool_call_id"] = json!(tc_id);
+            }
+        } else if m.role == "assistant" && !m.tool_calls.is_empty() {
+            let calls: Vec<serde_json::Value> = m
+                .tool_calls
+                .iter()
+                .map(|tc| {
+                    json!({
+                        "id": tc.id,
+                        "type": "function",
+                        "function": {
+                            "name": tc.name,
+                            "arguments": tc.arguments.to_string()
+                        }
+                    })
+                })
+                .collect();
+            msg["tool_calls"] = json!(calls);
+        }
+        msg
+    }));
+
+    let tool_json: Vec<serde_json::Value> = tools.iter().map(|t| t.to_openai_tool()).collect();
+
+    let mut body = json!({
+        "model": request.config.model,
+        "messages": messages,
+        "max_tokens": request.max_tokens.unwrap_or(4096),
+        "tools": tool_json,
+        "tool_choice": "auto",
+        "stream": true,
+    });
+    add_temperature_if_supported(&mut body, request);
+
+    let res = client
+        .post(resolve_endpoint(&request.config))
+        .headers(headers)
+        .json(&body)
+        .send()
+        .await
+        .map_err(|e| format!("AI request failed: {e}"))?;
+
+    if !res.status().is_success() {
+        let data: serde_json::Value = res.json().await.map_err(|e| e.to_string())?;
+        return Err(extract_error(&data).unwrap_or_else(|| "API error".to_string()));
+    }
+
+    let mut byte_stream = res.bytes_stream();
+    let mut buf = String::new();
+
+    loop {
+        tokio::select! {
+            chunk = byte_stream.next() => {
+                let Some(chunk) = chunk else { break };
+                let chunk = chunk.map_err(|e| e.to_string())?;
+                buf.push_str(&String::from_utf8_lossy(&chunk));
+
+                let mut finished = false;
+                while let Some(pos) = buf.find('\n') {
+                    let line = buf[..pos].to_string();
+                    buf = buf[pos + 1..].to_string();
+
+                    let Some(data) = stream_data_payload(&line) else { continue };
+                    if data == "[DONE]" {
+                        finished = true;
+                        break;
+                    }
+
+                    if let Ok(event) = serde_json::from_str::<serde_json::Value>(data) {
+                        // Reasoning
+                        if let Some(reasoning) = openai_stream_reasoning(&event) {
+                            on_event(StreamToolEvent::Chunk(AiStreamChunk {
+                                session_id: session_id.to_string(),
+                                delta: String::new(),
+                                reasoning_delta: Some(reasoning.to_string()),
+                                done: false,
+                            }));
+                        }
+                        // Text
+                        if let Some(text) = openai_stream_text(&event) {
+                            on_event(StreamToolEvent::Chunk(AiStreamChunk {
+                                session_id: session_id.to_string(),
+                                delta: text,
+                                reasoning_delta: None,
+                                done: false,
+                            }));
+                        }
+                        // Tool calls
+                        if let Some(tool_calls) = event["choices"].get(0).and_then(|c| c["delta"]["tool_calls"].as_array()) {
+                            for tc in tool_calls {
+                                let idx = tc["index"].as_u64().unwrap_or(0) as u32;
+                                // First chunk for this tool call has id and name
+                                if let Some(id) = tc["id"].as_str() {
+                                    let name = tc["function"]["name"].as_str().unwrap_or_default().to_string();
+                                    on_event(StreamToolEvent::ToolCallStart { index: idx, id: id.to_string(), name });
+                                }
+                                // Argument fragments
+                                if let Some(fragment) = tc["function"]["arguments"].as_str() {
+                                    on_event(StreamToolEvent::ToolCallDelta { index: idx, fragment: fragment.to_string() });
+                                }
+                            }
+                        }
+                    }
+                }
+
+                if finished { break; }
+            }
+            _ = cancelled.notified() => { break; }
+        }
+    }
+
+    Ok(())
+}
+
+/// Streaming Gemini call with tool support.
+async fn stream_gemini_with_tools(
+    client: &reqwest::Client,
+    session_id: &str,
+    request: &AiCompletionRequest,
+    tools: &[crate::agent_events::ToolDefinition],
+    cancelled: &Notify,
+    on_event: &impl Fn(StreamToolEvent),
+) -> Result<(), String> {
+    let mut contents: Vec<serde_json::Value> = Vec::new();
+    let mut pending_function_responses: Vec<serde_json::Value> = Vec::new();
+    for m in &request.messages {
+        if m.role == "tool" {
+            let tool_name = m
+                .tool_call_id
+                .as_deref()
+                .and_then(|s| s.strip_prefix("gemini-tc-"))
+                .and_then(|s| s.rsplitn(2, '-').nth(1))
+                .unwrap_or("unknown");
+            pending_function_responses.push(json!({
+                "functionResponse": {
+                    "name": tool_name,
+                    "response": { "content": m.content }
+                }
+            }));
+        } else {
+            // Flush any pending function responses before emitting a non-tool message.
+            if !pending_function_responses.is_empty() {
+                contents.push(json!({
+                    "role": "user",
+                    "parts": pending_function_responses.drain(..).collect::<Vec<_>>()
+                }));
+            }
+            if m.role == "assistant" && !m.tool_calls.is_empty() {
+                let mut parts: Vec<serde_json::Value> = Vec::new();
+                if !m.content.is_empty() {
+                    parts.push(json!({ "text": m.content }));
+                }
+                for tc in &m.tool_calls {
+                    parts.push(json!({ "functionCall": { "name": tc.name, "args": tc.arguments } }));
+                }
+                contents.push(json!({ "role": "model", "parts": parts }));
+            } else {
+                let role = if m.role == "assistant" { "model" } else { "user" };
+                contents.push(json!({ "role": role, "parts": [{ "text": m.content }] }));
+            }
+        }
+    }
+    // Flush any remaining function responses at the end of the message list.
+    if !pending_function_responses.is_empty() {
+        contents.push(json!({
+            "role": "user",
+            "parts": pending_function_responses.drain(..).collect::<Vec<_>>()
+        }));
+    }
+
+    let tool_declarations: Vec<serde_json::Value> = tools.iter().map(|t| t.to_gemini_tool()).collect();
+
+    let body = json!({
+        "contents": contents,
+        "systemInstruction": { "parts": [{ "text": request.system_prompt }] },
+        "tools": [{ "functionDeclarations": tool_declarations }],
+        "generationConfig": {
+            "maxOutputTokens": request.max_tokens.unwrap_or(4096),
+        }
+    });
+
+    let res = client
+        .post(resolve_gemini_stream_endpoint(&request.config))
+        .query(&[("key", request.config.api_key.as_str()), ("alt", "sse")])
+        .header(CONTENT_TYPE, "application/json")
+        .json(&body)
+        .send()
+        .await
+        .map_err(|e| format!("Gemini request failed: {e}"))?;
+
+    if !res.status().is_success() {
+        let data: serde_json::Value = res.json().await.map_err(|e| e.to_string())?;
+        return Err(extract_error(&data).unwrap_or_else(|| "Gemini API error".to_string()));
+    }
+
+    let mut byte_stream = res.bytes_stream();
+    let mut buf = String::new();
+    let mut tool_call_idx: u32 = 0;
+
+    loop {
+        tokio::select! {
+            chunk = byte_stream.next() => {
+                let Some(chunk) = chunk else { break };
+                let chunk = chunk.map_err(|e| e.to_string())?;
+                buf.push_str(&String::from_utf8_lossy(&chunk));
+
+                while let Some(pos) = buf.find('\n') {
+                    let line = buf[..pos].to_string();
+                    buf = buf[pos + 1..].to_string();
+
+                    let Some(data) = stream_data_payload(&line) else { continue };
+                    if let Ok(event) = serde_json::from_str::<serde_json::Value>(data) {
+                        if let Some(candidates) = event["candidates"].as_array() {
+                            if let Some(parts) = candidates[0]["content"]["parts"].as_array() {
+                                for part in parts {
+                                    // Text
+                                    if let Some(text) = part["text"].as_str() {
+                                        on_event(StreamToolEvent::Chunk(AiStreamChunk {
+                                            session_id: session_id.to_string(),
+                                            delta: text.to_string(),
+                                            reasoning_delta: None,
+                                            done: false,
+                                        }));
+                                    }
+                                    // Function call (Gemini sends complete objects, not deltas)
+                                    if let Some(fc) = part.get("functionCall") {
+                                        let name = fc["name"].as_str().unwrap_or_default().to_string();
+                                        let args = fc["args"].clone();
+                                        let id = format!("gemini-tc-{name}-{tool_call_idx}");
+                                        let args_str = args.to_string();
+                                        on_event(StreamToolEvent::ToolCallStart {
+                                            index: tool_call_idx,
+                                            id: id.clone(),
+                                            name: name.clone(),
+                                        });
+                                        on_event(StreamToolEvent::ToolCallDelta {
+                                            index: tool_call_idx,
+                                            fragment: args_str,
+                                        });
+                                        on_event(StreamToolEvent::ToolCallComplete { index: tool_call_idx });
+                                        tool_call_idx += 1;
+                                    }
+                                }
+                            }
+                        }
+                    }
+                }
+            }
+            _ = cancelled.notified() => { break; }
+        }
+    }
+
+    Ok(())
+}
+
+/// Public entry point: stream an LLM call with tool support, accumulating tool calls.
+/// Returns completed tool calls when the stream finishes.
+pub async fn stream_with_tools(
+    config: &AiConfig,
+    request: &AiCompletionRequest,
+    session_id: &str,
+    tools: &[crate::agent_events::ToolDefinition],
+    cancelled: &Notify,
+    on_chunk: impl Fn(AiStreamChunk),
+) -> Result<Vec<crate::agent_events::ToolCall>, String> {
+    validate_config(config)?;
+
+    let stream_timeout = if config.enable_thinking { 600 } else { 120 };
+    let client = build_ai_http_client(config, stream_timeout)?;
+
+    let accumulator = Arc::new(std::sync::Mutex::new(StreamingToolCallAccumulator::new()));
+
+    match config.provider {
+        AiProvider::Claude => {
+            stream_claude_with_tools(&client, session_id, request, tools, cancelled, &|event| {
+                accumulator.lock().unwrap().process(event, &on_chunk);
+            })
+            .await?
+        }
+        AiProvider::Gemini => {
+            stream_gemini_with_tools(&client, session_id, request, tools, cancelled, &|event| {
+                accumulator.lock().unwrap().process(event, &on_chunk);
+            })
+            .await?
+        }
+        _ => {
+            stream_openai_with_tools(&client, session_id, request, tools, cancelled, &|event| {
+                accumulator.lock().unwrap().process(event, &on_chunk);
+            })
+            .await?
+        }
+    }
+
+    Ok(Arc::try_unwrap(accumulator)
+        .expect("stream_with_tools: accumulator Arc should have single owner")
+        .into_inner()
+        .expect("stream_with_tools: accumulator Mutex should not be poisoned")
+        .finalize())
+}
+
 // ---------------------------------------------------------------------------
 // Conversation persistence (path-based)
 // ---------------------------------------------------------------------------
diff --git a/crates/dbx-core/src/db/postgres.rs b/crates/dbx-core/src/db/postgres.rs
index 2031454f91..ac79960bb1 100644
--- a/crates/dbx-core/src/db/postgres.rs
+++ b/crates/dbx-core/src/db/postgres.rs
@@ -1114,6 +1114,7 @@ pub async fn list_databases(pool: &Pool) -> Result<Vec<DatabaseInfo>, String> {
 }
 
 pub async fn list_tables(pool: &Pool, schema: &str) -> Result<Vec<TableInfo>, String> {
+    let schema = if schema.is_empty() { "public" } else { schema };
     let client = pool.get().await.map_err(|e| e.to_string())?;
     let stmt = client.prepare_cached(postgres_tables_sql()).await.map_err(|e| e.to_string())?;
     let rows = client.query(&stmt, &[&schema]).await.map_err(|e| e.to_string())?;
@@ -1492,6 +1493,7 @@ async fn get_columns_with_sql(
 }
 
 pub async fn get_columns(pool: &Pool, schema: &str, table: &str) -> Result<Vec<ColumnInfo>, String> {
+    let schema = if schema.is_empty() { "public" } else { schema };
     let client = pool.get().await.map_err(|e| e.to_string())?;
     match get_columns_with_sql(&client, POSTGRES_COLUMNS_SQL, schema, table).await {
         Ok(columns) => Ok(columns),
diff --git a/crates/dbx-core/src/lib.rs b/crates/dbx-core/src/lib.rs
index 1e2c22e4ce..b000024b46 100644
--- a/crates/dbx-core/src/lib.rs
+++ b/crates/dbx-core/src/lib.rs
@@ -1,9 +1,12 @@
 pub mod agent_catalog;
 pub mod agent_connection;
+pub mod agent_events;
 pub mod agent_kv;
+pub mod agent_loop;
 pub mod agent_manager;
 pub mod agent_runtime;
 pub mod agent_service;
+pub mod agent_tools;
 pub mod ai;
 pub mod cloud_sync;
 pub mod connection;
diff --git a/crates/dbx-core/src/schema/providers/native.rs b/crates/dbx-core/src/schema/providers/native.rs
index 816e6cb27a..013e3a6fa9 100644
--- a/crates/dbx-core/src/schema/providers/native.rs
+++ b/crates/dbx-core/src/schema/providers/native.rs
@@ -35,8 +35,14 @@ pub(in crate::schema) async fn list_tables(
         PoolKind::Mysql(p, _) if config.is_some_and(is_doris_family_config) => {
             db::mysql::list_tables_show(p, database).await
         }
-        PoolKind::Mysql(p, mode) if *mode == MysqlMode::OceanBaseOracle => db::ob_oracle::list_tables(p, schema).await,
-        PoolKind::Mysql(p, _) => db::mysql::list_tables(p, schema).await,
+        PoolKind::Mysql(p, mode) if *mode == MysqlMode::OceanBaseOracle => {
+            let s = if schema.is_empty() { database } else { schema };
+            db::ob_oracle::list_tables(p, s).await
+        }
+        PoolKind::Mysql(p, _) => {
+            let db = if schema.is_empty() { database } else { schema };
+            db::mysql::list_tables(p, db).await
+        }
         PoolKind::Postgres(p) => db::postgres::list_tables(p, schema).await,
         PoolKind::Sqlite(p) => db::sqlite::list_tables(p, schema).await,
         PoolKind::Rqlite(client) => db::rqlite_driver::list_tables(client, schema).await,
diff --git a/crates/dbx-web/src/main.rs b/crates/dbx-web/src/main.rs
index 47cc0718cb..50df7aa7db 100644
--- a/crates/dbx-web/src/main.rs
+++ b/crates/dbx-web/src/main.rs
@@ -290,6 +290,7 @@ async fn main() {
         .route("/ai/conversation/{id}", delete(routes::ai::delete_ai_conversation))
         .route("/ai/complete", post(routes::ai::ai_complete))
         .route("/ai/stream", post(routes::ai::ai_stream))
+        .route("/ai/agent-stream", post(routes::ai::ai_agent_stream))
         .route("/ai/cancel-stream", post(routes::ai::ai_cancel_stream))
         .route("/ai/test-connection", post(routes::ai::ai_test_connection))
         .route("/ai/models", post(routes::ai::ai_list_models))
diff --git a/crates/dbx-web/src/routes/ai.rs b/crates/dbx-web/src/routes/ai.rs
index 5a0ba87de5..c983bea941 100644
--- a/crates/dbx-web/src/routes/ai.rs
+++ b/crates/dbx-web/src/routes/ai.rs
@@ -6,7 +6,10 @@ use axum::Json;
 use futures::stream::Stream;
 use serde::Deserialize;
 
+use dbx_core::agent_events::AgentEvent;
+use dbx_core::agent_loop::{run_agent_loop, AgentLoopContext};
 use dbx_core::ai::{AiCompletionRequest, AiConfig, AiConversation, AiModelInfo, AiStreamChunk};
+use dbx_core::models::connection::DatabaseType;
 
 use crate::error::AppError;
 use crate::state::WebState;
@@ -58,6 +61,16 @@ pub struct AiCancelStreamRequest {
     pub session_id: String,
 }
 
+#[derive(Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct AiAgentStreamRequest {
+    pub session_id: String,
+    pub request: AiCompletionRequest,
+    pub connection_id: String,
+    pub database: String,
+    pub db_type: String,
+}
+
 // ---------------------------------------------------------------------------
 // Config
 // ---------------------------------------------------------------------------
@@ -171,3 +184,72 @@ pub async fn ai_stream(
 
     Ok(Sse::new(stream).keep_alive(KeepAlive::default()))
 }
+
+// ---------------------------------------------------------------------------
+// AI agent stream (POST returns SSE with AgentEvent)
+// ---------------------------------------------------------------------------
+
+pub async fn ai_agent_stream(
+    State(state): State<Arc<WebState>>,
+    Json(body): Json<AiAgentStreamRequest>,
+) -> Result<Sse<impl Stream<Item = Result<Event, std::convert::Infallible>>>, AppError> {
+    let session_id = body.session_id;
+    let request = body.request;
+
+    let cancelled = dbx_core::ai::register_stream(&session_id).await;
+    let (tx, rx) = tokio::sync::broadcast::channel::<String>(256);
+
+    let parsed_db_type: DatabaseType = serde_json::from_str(&format!("\"{}\"", body.db_type))
+        .map_err(|_| AppError(format!("Unknown database type: {}", body.db_type)))?;
+
+    let agent_ctx = AgentLoopContext {
+        state: state.app.clone(),
+        connection_id: body.connection_id,
+        database: body.database,
+        db_type: parsed_db_type,
+    };
+
+    let sid = session_id.clone();
+    let req_config = request.config;
+    let req_system_prompt = request.system_prompt;
+    let req_messages = request.messages;
+    let req_max_tokens = request.max_tokens;
+    let req_temperature = request.temperature;
+    let tx2 = tx.clone();
+    tokio::task::spawn_blocking(move || {
+        let rt =
+            tokio::runtime::Builder::new_current_thread().enable_all().build().expect("failed to create agent runtime");
+        rt.block_on(async move {
+            let result = run_agent_loop(
+                &req_config,
+                &req_system_prompt,
+                &req_messages,
+                &agent_ctx,
+                move |event: AgentEvent| {
+                    let json = serde_json::to_string(&event).unwrap_or_default();
+                    let _ = tx2.send(json);
+                },
+                &cancelled,
+                req_max_tokens,
+                req_temperature,
+            )
+            .await;
+
+            if let Err(e) = result {
+                let error_event = AgentEvent::Error { message: e };
+                let _ = tx.send(serde_json::to_string(&error_event).unwrap_or_default());
+            }
+
+            dbx_core::ai::unregister_stream(&sid).await;
+        });
+    });
+
+    let stream = async_stream::stream! {
+        let mut rx = rx;
+        while let Ok(data) = rx.recv().await {
+            yield Ok(Event::default().data(data));
+        }
+    };
+
+    Ok(Sse::new(stream).keep_alive(KeepAlive::default()))
+}
diff --git a/src-tauri/src/commands/ai.rs b/src-tauri/src/commands/ai.rs
index 9439eb9131..b048b04691 100644
--- a/src-tauri/src/commands/ai.rs
+++ b/src-tauri/src/commands/ai.rs
@@ -42,11 +42,53 @@ pub async fn ai_stream(app: AppHandle, session_id: String, request: AiCompletion
     result
 }
 
+use dbx_core::agent_events::AgentEvent;
+use dbx_core::agent_loop::{run_agent_loop, AgentLoopContext};
+use dbx_core::models::connection::DatabaseType;
+
 #[tauri::command]
 pub async fn ai_cancel_stream(session_id: String) -> Result<bool, String> {
     Ok(dbx_core::ai::cancel_stream(&session_id).await)
 }
 
+#[tauri::command]
+pub async fn ai_agent_stream(
+    app: AppHandle,
+    state: State<'_, Arc<AppState>>,
+    session_id: String,
+    request: AiCompletionRequest,
+    connection_id: String,
+    database: String,
+    db_type: String,
+) -> Result<String, String> {
+    let cancelled = dbx_core::ai::register_stream(&session_id).await;
+
+    let parsed_db_type: DatabaseType =
+        serde_json::from_str(&format!("\"{}\"", db_type)).map_err(|_| format!("Unknown database type: {db_type}"))?;
+
+    let agent_ctx = AgentLoopContext { state: state.inner().clone(), connection_id, database, db_type: parsed_db_type };
+
+    let result = run_agent_loop(
+        &request.config,
+        &request.system_prompt,
+        &request.messages,
+        &agent_ctx,
+        {
+            let app = app.clone();
+            move |event: AgentEvent| {
+                let _ = app.emit("ai-agent-event", &event);
+            }
+        },
+        &cancelled,
+        request.max_tokens,
+        request.temperature,
+    )
+    .await;
+
+    dbx_core::ai::unregister_stream(&session_id).await;
+    result
+}
+
 #[tauri::command]
 pub async fn save_ai_conversation(state: State<'_, Arc<AppState>>, conversation: AiConversation) -> Result<(), String> {
     state.storage.save_ai_conversation(&conversation).await
diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs
index 6b83075111..ac40d114a6 100644
--- a/src-tauri/src/lib.rs
+++ b/src-tauri/src/lib.rs
@@ -317,6 +317,7 @@ pub fn run() {
         .invoke_handler(tauri::generate_handler![
             commands::ai::ai_complete,
             commands::ai::ai_stream,
+            commands::ai::ai_agent_stream,
             commands::ai::ai_cancel_stream,
             commands::ai::ai_test_connection,
             commands::ai::ai_list_models,