feat: add datatable read and write MCP tools

backend/Cargo.lock

@@ -14473,6 +14473,7 @@ dependencies = [
  "sqlx",
  "strum",
  "tokio",
+ "tokio-postgres",
  "tracing",
  "uuid",
  "windmill-api-auth",

backend/windmill-api-workspaces/Cargo.toml

@@ -47,6 +47,7 @@ serde_json.workspace = true
 sha2.workspace = true
 sqlx.workspace = true
 tokio.workspace = true
+tokio-postgres.workspace = true
 tracing.workspace = true
 uuid.workspace = true
 strum.workspace = true

backend/windmill-api-workspaces/src/workspaces.rs

@@ -131,6 +131,9 @@ pub fn workspaced_service() -> Router {
             "/get_datatable_table_schema",
             get(get_datatable_table_schema),
         )
+        .route("/query_datatable", get(query_datatable))
+        .route("/insert_datatable", post(insert_datatable))
+        .route("/update_datatable", post(update_datatable))
         .route("/edit_datatable_config", post(edit_datatable_config))
         .route("/git_sync_enabled", get(get_git_sync_enabled))
         .route("/edit_git_sync_config", post(edit_git_sync_config))
@@ -1875,6 +1878,261 @@ fn is_system_pg_schema(schema_name: &str) -> bool {
     ) || schema_name.starts_with("pg_")
 }
 
+// ---------------------------------------------------------------------------
+// Datatable data operations (read/write) — exposed as MCP tools.
+//
+// These connect directly to the datatable's underlying Postgres (same idiom as
+// `get_datatable_schema`) and run a single, structurally-fixed statement. Only
+// the SELECT/INSERT/UPDATE shape is ours; identifiers are always quoted and
+// values are bound through `jsonb_populate_record` so Postgres does the type
+// coercion, never string interpolation. A caller-supplied `where_clause` is the
+// one raw fragment — it is trusted like the DB-manager's predicate, and because
+// `client.query`/`execute` prepare a single statement, it cannot chain a second
+// one.
+// ---------------------------------------------------------------------------
+
+const DATATABLE_QUERY_MAX_LIMIT: i64 = 1000;
+const DATATABLE_QUERY_DEFAULT_LIMIT: i64 = 100;
+
+fn default_public_schema() -> String {
+    "public".to_string()
+}
+
+/// Connect to a datatable's underlying Postgres, spawning the connection driver.
+async fn connect_datatable(
+    db: &DB,
+    w_id: &str,
+    datatable_name: &str,
+) -> Result<tokio_postgres::Client> {
+    let db_resource = get_datatable_resource_from_db_unchecked(db, w_id, datatable_name).await?;
+    let pg_db: PgDatabase = serde_json::from_value(db_resource)
+        .map_err(|e| Error::internal_err(format!("Failed to parse database credentials: {}", e)))?;
+    let (client, connection) = pg_db.connect(Some(db)).await?;
+    tokio::spawn(async move {
+        if let Err(e) = connection.await {
+            tracing::error!("Datatable connection error: {}", e);
+        }
+    });
+    Ok(client)
+}
+
+/// Quote a Postgres identifier by doubling embedded quotes. Rejects empty
+/// identifiers and NUL bytes so a column/table/schema name can never break out
+/// of the quotes.
+fn quote_pg_ident(ident: &str) -> Result<String> {
+    if ident.is_empty() || ident.contains('\0') {
+        return Err(Error::BadRequest(format!(
+            "Invalid SQL identifier: {:?}",
+            ident
+        )));
+    }
+    Ok(format!("\"{}\"", ident.replace('"', "\"\"")))
+}
+
+fn check_datatable_schema(schema_name: &str) -> Result<()> {
+    if is_system_pg_schema(schema_name) {
+        return Err(Error::BadRequest(format!(
+            "Schema '{}' is a system schema and cannot be used for datatable data operations",
+            schema_name
+        )));
+    }
+    Ok(())
+}
+
+#[derive(Deserialize)]
+struct QueryDataTableQuery {
+    datatable_name: String,
+    #[serde(default = "default_public_schema")]
+    schema_name: String,
+    table_name: String,
+    /// Raw SQL predicate AND-ed into the SELECT (e.g. `status = 'active' AND age > 18`).
+    where_clause: Option<String>,
+    /// Comma-separated columns to return. Defaults to all columns.
+    select: Option<String>,
+    /// Raw ORDER BY expression (e.g. `created_at DESC`).
+    order_by: Option<String>,
+    limit: Option<i64>,
+    offset: Option<i64>,
+}
+
+/// SELECT rows from a datatable table with an optional WHERE predicate.
+async fn query_datatable(
+    _authed: ApiAuthed,
+    Extension(db): Extension<DB>,
+    Path(w_id): Path<String>,
+    Query(q): Query<QueryDataTableQuery>,
+) -> JsonResult<Vec<serde_json::Value>> {
+    check_datatable_schema(&q.schema_name)?;
+
+    let limit = q
+        .limit
+        .unwrap_or(DATATABLE_QUERY_DEFAULT_LIMIT)
+        .clamp(1, DATATABLE_QUERY_MAX_LIMIT);
+    let offset = q.offset.unwrap_or(0).max(0);
+
+    let qualified = format!(
+        "{}.{}",
+        quote_pg_ident(&q.schema_name)?,
+        quote_pg_ident(&q.table_name)?
+    );
+
+    let select_list = match q.select.as_ref().filter(|s| !s.trim().is_empty()) {
+        Some(cols) => cols
+            .split(',')
+            .map(|c| quote_pg_ident(c.trim()))
+            .collect::<Result<Vec<_>>>()?
+            .join(", "),
+        None => "*".to_string(),
+    };
+
+    let mut inner = format!("SELECT {} FROM {}", select_list, qualified);
+    if let Some(w) = q.where_clause.as_ref().filter(|w| !w.trim().is_empty()) {
+        inner.push_str(&format!(" WHERE {}", w));
+    }
+    if let Some(o) = q.order_by.as_ref().filter(|o| !o.trim().is_empty()) {
+        inner.push_str(&format!(" ORDER BY {}", o));
+    }
+    inner.push_str(&format!(" LIMIT {} OFFSET {}", limit, offset));
+
+    let sql = format!("SELECT to_jsonb(_wm_row) AS row FROM ({}) _wm_row", inner);
+
+    let client = connect_datatable(&db, &w_id, &q.datatable_name).await?;
+    let rows = client
+        .query(&sql, &[])
+        .await
+        .map_err(|e| Error::BadRequest(format!("Datatable query failed: {}", e)))?;
+
+    let out = rows
+        .into_iter()
+        .map(|r| r.get::<_, serde_json::Value>(0))
+        .collect();
+    Ok(Json(out))
+}
+
+#[derive(Deserialize)]
+struct InsertDataTableRequest {
+    datatable_name: String,
+    #[serde(default = "default_public_schema")]
+    schema_name: String,
+    table_name: String,
+    /// Column name -> value. Types are coerced by Postgres from the JSON object;
+    /// columns not listed keep their table default.
+    values: serde_json::Map<String, serde_json::Value>,
+}
+
+/// INSERT a single row into a datatable table, returning the inserted row.
+async fn insert_datatable(
+    _authed: ApiAuthed,
+    Extension(db): Extension<DB>,
+    Path(w_id): Path<String>,
+    Json(req): Json<InsertDataTableRequest>,
+) -> JsonResult<serde_json::Value> {
+    check_datatable_schema(&req.schema_name)?;
+    if req.values.is_empty() {
+        return Err(Error::BadRequest(
+            "`values` must contain at least one column".to_string(),
+        ));
+    }
+
+    let qualified = format!(
+        "{}.{}",
+        quote_pg_ident(&req.schema_name)?,
+        quote_pg_ident(&req.table_name)?
+    );
+
+    let cols = req
+        .values
+        .keys()
+        .map(|k| quote_pg_ident(k))
+        .collect::<Result<Vec<_>>>()?;
+    let col_list = cols.join(", ");
+    let select_list = cols.join(", ");
+
+    let json_obj = serde_json::Value::Object(req.values.clone());
+    let sql = format!(
+        "INSERT INTO {qualified} AS _wm_t ({col_list}) \
+         SELECT {select_list} FROM jsonb_populate_record(NULL::{qualified}, $1::jsonb) \
+         RETURNING to_jsonb(_wm_t)"
+    );
+
+    let client = connect_datatable(&db, &w_id, &req.datatable_name).await?;
+    let row = client
+        .query_one(&sql, &[&json_obj])
+        .await
+        .map_err(|e| Error::BadRequest(format!("Datatable insert failed: {}", e)))?;
+
+    Ok(Json(row.get::<_, serde_json::Value>(0)))
+}
+
+#[derive(Deserialize)]
+struct UpdateDataTableRequest {
+    datatable_name: String,
+    #[serde(default = "default_public_schema")]
+    schema_name: String,
+    table_name: String,
+    /// Column name -> new value. Types are coerced by Postgres from the JSON object.
+    set: serde_json::Map<String, serde_json::Value>,
+    /// Raw SQL predicate selecting which rows to update. Required to prevent an
+    /// accidental full-table update.
+    where_clause: String,
+}
+
+/// UPDATE rows of a datatable table matching a WHERE predicate. Returns the
+/// number of updated rows.
+async fn update_datatable(
+    _authed: ApiAuthed,
+    Extension(db): Extension<DB>,
+    Path(w_id): Path<String>,
+    Json(req): Json<UpdateDataTableRequest>,
+) -> JsonResult<serde_json::Value> {
+    check_datatable_schema(&req.schema_name)?;
+    if req.set.is_empty() {
+        return Err(Error::BadRequest(
+            "`set` must contain at least one column".to_string(),
+        ));
+    }
+    if req.where_clause.trim().is_empty() {
+        return Err(Error::BadRequest(
+            "`where_clause` is required to avoid updating the whole table".to_string(),
+        ));
+    }
+
+    let qualified = format!(
+        "{}.{}",
+        quote_pg_ident(&req.schema_name)?,
+        quote_pg_ident(&req.table_name)?
+    );
+
+    // Each SET value is pulled as a scalar out of a per-column
+    // `jsonb_populate_record` so there is no FROM-join — the WHERE predicate's
+    // bare column names then resolve unambiguously to the target table.
+    let set_clause = req
+        .set
+        .keys()
+        .map(|k| {
+            let col = quote_pg_ident(k)?;
+            Ok(format!(
+                "{col} = (jsonb_populate_record(NULL::{qualified}, $1::jsonb)).{col}"
+            ))
+        })
+        .collect::<Result<Vec<_>>>()?
+        .join(", ");
+
+    let json_obj = serde_json::Value::Object(req.set.clone());
+    let sql = format!(
+        "UPDATE {qualified} SET {set_clause} WHERE {}",
+        req.where_clause
+    );
+
+    let client = connect_datatable(&db, &w_id, &req.datatable_name).await?;
+    let updated = client
+        .execute(&sql, &[&json_obj])
+        .await
+        .map_err(|e| Error::BadRequest(format!("Datatable update failed: {}", e)))?;
+
+    Ok(Json(serde_json::json!({ "updated": updated })))
+}
+
 fn compact_column_type(
     udt_name: String,
     is_nullable: String,
@@ -1921,6 +2179,24 @@ mod tests {
             format!("text={}...", "é".repeat(27))
         );
     }
+
+    #[test]
+    fn quote_pg_ident_wraps_and_escapes() {
+        assert_eq!(quote_pg_ident("id").unwrap(), "\"id\"");
+        assert_eq!(quote_pg_ident("my Table").unwrap(), "\"my Table\"");
+        // A caller injecting a closing quote must be neutralized by doubling it,
+        // not by breaking out of the quotes.
+        assert_eq!(
+            quote_pg_ident("a\"; DROP TABLE t; --").unwrap(),
+            "\"a\"\"; DROP TABLE t; --\""
+        );
+    }
+
+    #[test]
+    fn quote_pg_ident_rejects_empty_and_nul() {
+        assert!(quote_pg_ident("").is_err());
+        assert!(quote_pg_ident("a\0b").is_err());
+    }
 }
 
 /// Resolve a source string to PgDatabase credentials with user-scoped permission checks.

backend/windmill-api/openapi.yaml

@@ -4495,6 +4495,8 @@ paths:
     get:
       summary: list Datatables
       operationId: listDataTables
+      x-mcp-tool: true
+      x-mcp-instructions: "Use this first to discover which data tables (named Postgres databases) exist in the workspace. Returns each data table's name; pass that name as `datatable_name` to the other data table tools."
       tags:
         - workspace
       parameters:
@@ -4540,6 +4542,8 @@ paths:
     get:
       summary: list tables of all connected Datatables
       operationId: listDataTableTables
+      x-mcp-tool: true
+      x-mcp-instructions: "Lists, for every data table in the workspace, its schemas and the tables inside them. Use this to find which tables you can read from or write to before calling getDataTableTableSchema or queryDataTable."
       tags:
         - workspace
       parameters:
@@ -4558,6 +4562,8 @@ paths:
     get:
       summary: get one Datatable table schema
       operationId: getDataTableTableSchema
+      x-mcp-tool: true
+      x-mcp-instructions: "Returns the columns of a single data table table, each with its Postgres type, nullability and default. Call this before queryDataTable/insertDataTable/updateDataTable so you know the exact column names and types."
       tags:
         - workspace
       parameters:
@@ -4585,6 +4591,158 @@ paths:
               schema:
                 $ref: "#/components/schemas/DataTableTableSchema"
 
+  /w/{workspace}/workspaces/query_datatable:
+    get:
+      summary: query rows from a datatable table
+      operationId: queryDataTable
+      x-mcp-tool: true
+      x-mcp-instructions: "Read rows from a data table table with an optional WHERE predicate. `where_clause` is a raw SQL boolean expression on the table's columns (e.g. `status = 'active' AND age > 18`); use single quotes for string literals. Only the columns you need should be listed in `select`. Results are capped at 1000 rows (default 100) — use `limit`, `offset` and `order_by` to page. Call getDataTableTableSchema first to learn the column names and types."
+      tags:
+        - workspace
+      parameters:
+        - $ref: "#/components/parameters/WorkspaceId"
+        - name: datatable_name
+          in: query
+          required: true
+          description: "Name of the data table (from listDataTables)"
+          schema:
+            type: string
+        - name: table_name
+          in: query
+          required: true
+          description: "Table to read from"
+          schema:
+            type: string
+        - name: schema_name
+          in: query
+          required: false
+          description: "Postgres schema of the table (defaults to `public`)"
+          schema:
+            type: string
+        - name: select
+          in: query
+          required: false
+          description: "Comma-separated list of columns to return. Defaults to all columns."
+          schema:
+            type: string
+        - name: where_clause
+          in: query
+          required: false
+          description: "Raw SQL predicate on the table's columns (the WHERE clause body, without the `WHERE` keyword)."
+          schema:
+            type: string
+        - name: order_by
+          in: query
+          required: false
+          description: "Raw SQL ORDER BY expression, e.g. `created_at DESC`."
+          schema:
+            type: string
+        - name: limit
+          in: query
+          required: false
+          description: "Maximum number of rows to return (1-1000, default 100)."
+          schema:
+            type: integer
+        - name: offset
+          in: query
+          required: false
+          description: "Number of rows to skip, for pagination."
+          schema:
+            type: integer
+      responses:
+        "200":
+          description: rows matching the query, each as a JSON object
+          content:
+            application/json:
+              schema:
+                type: array
+                items:
+                  type: object
+
+  /w/{workspace}/workspaces/insert_datatable:
+    post:
+      summary: insert a row into a datatable table
+      operationId: insertDataTable
+      x-mcp-tool: true
+      x-mcp-instructions: "Insert a single row into a data table table. `values` maps column names to values; Postgres coerces the JSON values to each column's type. Columns you omit keep their table default (so you can leave out serial/auto-generated primary keys). Returns the full inserted row. Call getDataTableTableSchema first to learn the column names and types."
+      tags:
+        - workspace
+      parameters:
+        - $ref: "#/components/parameters/WorkspaceId"
+      requestBody:
+        description: Row to insert
+        required: true
+        content:
+          application/json:
+            schema:
+              type: object
+              required: [datatable_name, table_name, values]
+              properties:
+                datatable_name:
+                  type: string
+                  description: "Name of the data table (from listDataTables)"
+                table_name:
+                  type: string
+                  description: "Table to insert into"
+                schema_name:
+                  type: string
+                  description: "Postgres schema of the table (defaults to `public`)"
+                values:
+                  type: object
+                  description: "Column name -> value for the row to insert"
+      responses:
+        "200":
+          description: the inserted row as a JSON object
+          content:
+            application/json:
+              schema:
+                type: object
+
+  /w/{workspace}/workspaces/update_datatable:
+    post:
+      summary: update rows of a datatable table
+      operationId: updateDataTable
+      x-mcp-tool: true
+      x-mcp-instructions: "Update rows of a data table table that match a WHERE predicate. `set` maps column names to their new values (types are coerced by Postgres). `where_clause` is a raw SQL predicate selecting which rows to update and is REQUIRED to prevent an accidental full-table update — target rows precisely (e.g. by primary key). Returns the number of updated rows. Call getDataTableTableSchema first to learn the column names and types."
+      tags:
+        - workspace
+      parameters:
+        - $ref: "#/components/parameters/WorkspaceId"
+      requestBody:
+        description: Update specification
+        required: true
+        content:
+          application/json:
+            schema:
+              type: object
+              required: [datatable_name, table_name, set, where_clause]
+              properties:
+                datatable_name:
+                  type: string
+                  description: "Name of the data table (from listDataTables)"
+                table_name:
+                  type: string
+                  description: "Table to update"
+                schema_name:
+                  type: string
+                  description: "Postgres schema of the table (defaults to `public`)"
+                set:
+                  type: object
+                  description: "Column name -> new value"
+                where_clause:
+                  type: string
+                  description: "Raw SQL predicate selecting the rows to update (required)"
+      responses:
+        "200":
+          description: number of updated rows
+          content:
+            application/json:
+              schema:
+                type: object
+                properties:
+                  updated:
+                    type: integer
+
   /w/{workspace}/workspaces/edit_ducklake_config:
     post:
       summary: edit ducklake settings

backend/windmill-api/src/mcp/auto_generated_endpoints.rs

@@ -58,6 +58,197 @@ pub fn all_tools() -> Vec<EndpointTool> {
         query_field_renames: None,
         body_field_renames: None,
     },
+    EndpointTool {
+        name: Cow::Borrowed("listDataTables"),
+        description: Cow::Borrowed("list Datatables"),
+        instructions: Cow::Borrowed("Use this first to discover which data tables (named Postgres databases) exist in the workspace. Returns each data table's name; pass that name as `datatable_name` to the other data table tools."),
+        path: Cow::Borrowed("/w/{workspace}/workspaces/list_datatables"),
+        method: Cow::Borrowed("GET"),
+        path_params_schema: None,
+        query_params_schema: None,
+        body_schema: None,
+        path_field_renames: None,
+        query_field_renames: None,
+        body_field_renames: None,
+    },
+    EndpointTool {
+        name: Cow::Borrowed("listDataTableTables"),
+        description: Cow::Borrowed("list tables of all connected Datatables"),
+        instructions: Cow::Borrowed("Lists, for every data table in the workspace, its schemas and the tables inside them. Use this to find which tables you can read from or write to before calling getDataTableTableSchema or queryDataTable."),
+        path: Cow::Borrowed("/w/{workspace}/workspaces/list_datatable_tables"),
+        method: Cow::Borrowed("GET"),
+        path_params_schema: None,
+        query_params_schema: None,
+        body_schema: None,
+        path_field_renames: None,
+        query_field_renames: None,
+        body_field_renames: None,
+    },
+    EndpointTool {
+        name: Cow::Borrowed("getDataTableTableSchema"),
+        description: Cow::Borrowed("get one Datatable table schema"),
+        instructions: Cow::Borrowed("Returns the columns of a single data table table, each with its Postgres type, nullability and default. Call this before queryDataTable/insertDataTable/updateDataTable so you know the exact column names and types."),
+        path: Cow::Borrowed("/w/{workspace}/workspaces/get_datatable_table_schema"),
+        method: Cow::Borrowed("GET"),
+        path_params_schema: None,
+        query_params_schema: Some(serde_json::json!({
+        "type": "object",
+        "properties": {
+                "datatable_name": {
+                        "type": "string"
+                },
+                "schema_name": {
+                        "type": "string"
+                },
+                "table_name": {
+                        "type": "string"
+                }
+        },
+        "required": [
+                "datatable_name",
+                "schema_name",
+                "table_name"
+        ]
+})),
+        body_schema: None,
+        path_field_renames: None,
+        query_field_renames: None,
+        body_field_renames: None,
+    },
+    EndpointTool {
+        name: Cow::Borrowed("queryDataTable"),
+        description: Cow::Borrowed("query rows from a datatable table"),
+        instructions: Cow::Borrowed("Read rows from a data table table with an optional WHERE predicate. `where_clause` is a raw SQL boolean expression on the table's columns (e.g. `status = 'active' AND age > 18`); use single quotes for string literals. Only the columns you need should be listed in `select`. Results are capped at 1000 rows (default 100) — use `limit`, `offset` and `order_by` to page. Call getDataTableTableSchema first to learn the column names and types."),
+        path: Cow::Borrowed("/w/{workspace}/workspaces/query_datatable"),
+        method: Cow::Borrowed("GET"),
+        path_params_schema: None,
+        query_params_schema: Some(serde_json::json!({
+        "type": "object",
+        "properties": {
+                "datatable_name": {
+                        "type": "string",
+                        "description": "Name of the data table (from listDataTables)"
+                },
+                "table_name": {
+                        "type": "string",
+                        "description": "Table to read from"
+                },
+                "schema_name": {
+                        "type": "string",
+                        "description": "Postgres schema of the table (defaults to `public`)"
+                },
+                "select": {
+                        "type": "string",
+                        "description": "Comma-separated list of columns to return. Defaults to all columns."
+                },
+                "where_clause": {
+                        "type": "string",
+                        "description": "Raw SQL predicate on the table's columns (the WHERE clause body, without the `WHERE` keyword)."
+                },
+                "order_by": {
+                        "type": "string",
+                        "description": "Raw SQL ORDER BY expression, e.g. `created_at DESC`."
+                },
+                "limit": {
+                        "type": "integer",
+                        "description": "Maximum number of rows to return (1-1000, default 100)."
+                },
+                "offset": {
+                        "type": "integer",
+                        "description": "Number of rows to skip, for pagination."
+                }
+        },
+        "required": [
+                "datatable_name",
+                "table_name"
+        ]
+})),
+        body_schema: None,
+        path_field_renames: None,
+        query_field_renames: None,
+        body_field_renames: None,
+    },
+    EndpointTool {
+        name: Cow::Borrowed("insertDataTable"),
+        description: Cow::Borrowed("insert a row into a datatable table"),
+        instructions: Cow::Borrowed("Insert a single row into a data table table. `values` maps column names to values; Postgres coerces the JSON values to each column's type. Columns you omit keep their table default (so you can leave out serial/auto-generated primary keys). Returns the full inserted row. Call getDataTableTableSchema first to learn the column names and types."),
+        path: Cow::Borrowed("/w/{workspace}/workspaces/insert_datatable"),
+        method: Cow::Borrowed("POST"),
+        path_params_schema: None,
+        query_params_schema: None,
+        body_schema: Some(serde_json::json!({
+        "type": "object",
+        "required": [
+                "datatable_name",
+                "table_name",
+                "values"
+        ],
+        "properties": {
+                "datatable_name": {
+                        "type": "string",
+                        "description": "Name of the data table (from listDataTables)"
+                },
+                "table_name": {
+                        "type": "string",
+                        "description": "Table to insert into"
+                },
+                "schema_name": {
+                        "type": "string",
+                        "description": "Postgres schema of the table (defaults to `public`)"
+                },
+                "values": {
+                        "type": "object",
+                        "description": "Column name -> value for the row to insert"
+                }
+        }
+})),
+        path_field_renames: None,
+        query_field_renames: None,
+        body_field_renames: None,
+    },
+    EndpointTool {
+        name: Cow::Borrowed("updateDataTable"),
+        description: Cow::Borrowed("update rows of a datatable table"),
+        instructions: Cow::Borrowed("Update rows of a data table table that match a WHERE predicate. `set` maps column names to their new values (types are coerced by Postgres). `where_clause` is a raw SQL predicate selecting which rows to update and is REQUIRED to prevent an accidental full-table update — target rows precisely (e.g. by primary key). Returns the number of updated rows. Call getDataTableTableSchema first to learn the column names and types."),
+        path: Cow::Borrowed("/w/{workspace}/workspaces/update_datatable"),
+        method: Cow::Borrowed("POST"),
+        path_params_schema: None,
+        query_params_schema: None,
+        body_schema: Some(serde_json::json!({
+        "type": "object",
+        "required": [
+                "datatable_name",
+                "table_name",
+                "set",
+                "where_clause"
+        ],
+        "properties": {
+                "datatable_name": {
+                        "type": "string",
+                        "description": "Name of the data table (from listDataTables)"
+                },
+                "table_name": {
+                        "type": "string",
+                        "description": "Table to update"
+                },
+                "schema_name": {
+                        "type": "string",
+                        "description": "Postgres schema of the table (defaults to `public`)"
+                },
+                "set": {
+                        "type": "object",
+                        "description": "Column name -> new value"
+                },
+                "where_clause": {
+                        "type": "string",
+                        "description": "Raw SQL predicate selecting the rows to update (required)"
+                }
+        }
+})),
+        path_field_renames: None,
+        query_field_renames: None,
+        body_field_renames: None,
+    },
     EndpointTool {
         name: Cow::Borrowed("createVariable"),
         description: Cow::Borrowed("create variable"),

frontend/src/lib/mcpEndpointTools.ts

@@ -68,6 +68,197 @@ export const mcpEndpointTools: EndpointTool[] = [
         queryFieldRenames: undefined,
         bodyFieldRenames: undefined
     },
+    {
+        name: "listDataTables",
+        description: "list Datatables",
+        instructions: "Use this first to discover which data tables (named Postgres databases) exist in the workspace. Returns each data table's name; pass that name as `datatable_name` to the other data table tools.",
+        path: "/w/{workspace}/workspaces/list_datatables",
+        method: "GET",
+        pathParamsSchema: undefined,
+        queryParamsSchema: undefined,
+        bodySchema: undefined,
+        pathFieldRenames: undefined,
+        queryFieldRenames: undefined,
+        bodyFieldRenames: undefined
+    },
+    {
+        name: "listDataTableTables",
+        description: "list tables of all connected Datatables",
+        instructions: "Lists, for every data table in the workspace, its schemas and the tables inside them. Use this to find which tables you can read from or write to before calling getDataTableTableSchema or queryDataTable.",
+        path: "/w/{workspace}/workspaces/list_datatable_tables",
+        method: "GET",
+        pathParamsSchema: undefined,
+        queryParamsSchema: undefined,
+        bodySchema: undefined,
+        pathFieldRenames: undefined,
+        queryFieldRenames: undefined,
+        bodyFieldRenames: undefined
+    },
+    {
+        name: "getDataTableTableSchema",
+        description: "get one Datatable table schema",
+        instructions: "Returns the columns of a single data table table, each with its Postgres type, nullability and default. Call this before queryDataTable/insertDataTable/updateDataTable so you know the exact column names and types.",
+        path: "/w/{workspace}/workspaces/get_datatable_table_schema",
+        method: "GET",
+        pathParamsSchema: undefined,
+        queryParamsSchema: {
+        "type": "object",
+        "properties": {
+                "datatable_name": {
+                        "type": "string"
+                },
+                "schema_name": {
+                        "type": "string"
+                },
+                "table_name": {
+                        "type": "string"
+                }
+        },
+        "required": [
+                "datatable_name",
+                "schema_name",
+                "table_name"
+        ]
+},
+        bodySchema: undefined,
+        pathFieldRenames: undefined,
+        queryFieldRenames: undefined,
+        bodyFieldRenames: undefined
+    },
+    {
+        name: "queryDataTable",
+        description: "query rows from a datatable table",
+        instructions: "Read rows from a data table table with an optional WHERE predicate. `where_clause` is a raw SQL boolean expression on the table's columns (e.g. `status = 'active' AND age > 18`); use single quotes for string literals. Only the columns you need should be listed in `select`. Results are capped at 1000 rows (default 100) — use `limit`, `offset` and `order_by` to page. Call getDataTableTableSchema first to learn the column names and types.",
+        path: "/w/{workspace}/workspaces/query_datatable",
+        method: "GET",
+        pathParamsSchema: undefined,
+        queryParamsSchema: {
+        "type": "object",
+        "properties": {
+                "datatable_name": {
+                        "type": "string",
+                        "description": "Name of the data table (from listDataTables)"
+                },
+                "table_name": {
+                        "type": "string",
+                        "description": "Table to read from"
+                },
+                "schema_name": {
+                        "type": "string",
+                        "description": "Postgres schema of the table (defaults to `public`)"
+                },
+                "select": {
+                        "type": "string",
+                        "description": "Comma-separated list of columns to return. Defaults to all columns."
+                },
+                "where_clause": {
+                        "type": "string",
+                        "description": "Raw SQL predicate on the table's columns (the WHERE clause body, without the `WHERE` keyword)."
+                },
+                "order_by": {
+                        "type": "string",
+                        "description": "Raw SQL ORDER BY expression, e.g. `created_at DESC`."
+                },
+                "limit": {
+                        "type": "integer",
+                        "description": "Maximum number of rows to return (1-1000, default 100)."
+                },
+                "offset": {
+                        "type": "integer",
+                        "description": "Number of rows to skip, for pagination."
+                }
+        },
+        "required": [
+                "datatable_name",
+                "table_name"
+        ]
+},
+        bodySchema: undefined,
+        pathFieldRenames: undefined,
+        queryFieldRenames: undefined,
+        bodyFieldRenames: undefined
+    },
+    {
+        name: "insertDataTable",
+        description: "insert a row into a datatable table",
+        instructions: "Insert a single row into a data table table. `values` maps column names to values; Postgres coerces the JSON values to each column's type. Columns you omit keep their table default (so you can leave out serial/auto-generated primary keys). Returns the full inserted row. Call getDataTableTableSchema first to learn the column names and types.",
+        path: "/w/{workspace}/workspaces/insert_datatable",
+        method: "POST",
+        pathParamsSchema: undefined,
+        queryParamsSchema: undefined,
+        bodySchema: {
+        "type": "object",
+        "required": [
+                "datatable_name",
+                "table_name",
+                "values"
+        ],
+        "properties": {
+                "datatable_name": {
+                        "type": "string",
+                        "description": "Name of the data table (from listDataTables)"
+                },
+                "table_name": {
+                        "type": "string",
+                        "description": "Table to insert into"
+                },
+                "schema_name": {
+                        "type": "string",
+                        "description": "Postgres schema of the table (defaults to `public`)"
+                },
+                "values": {
+                        "type": "object",
+                        "description": "Column name -> value for the row to insert"
+                }
+        }
+},
+        pathFieldRenames: undefined,
+        queryFieldRenames: undefined,
+        bodyFieldRenames: undefined
+    },
+    {
+        name: "updateDataTable",
+        description: "update rows of a datatable table",
+        instructions: "Update rows of a data table table that match a WHERE predicate. `set` maps column names to their new values (types are coerced by Postgres). `where_clause` is a raw SQL predicate selecting which rows to update and is REQUIRED to prevent an accidental full-table update — target rows precisely (e.g. by primary key). Returns the number of updated rows. Call getDataTableTableSchema first to learn the column names and types.",
+        path: "/w/{workspace}/workspaces/update_datatable",
+        method: "POST",
+        pathParamsSchema: undefined,
+        queryParamsSchema: undefined,
+        bodySchema: {
+        "type": "object",
+        "required": [
+                "datatable_name",
+                "table_name",
+                "set",
+                "where_clause"
+        ],
+        "properties": {
+                "datatable_name": {
+                        "type": "string",
+                        "description": "Name of the data table (from listDataTables)"
+                },
+                "table_name": {
+                        "type": "string",
+                        "description": "Table to update"
+                },
+                "schema_name": {
+                        "type": "string",
+                        "description": "Postgres schema of the table (defaults to `public`)"
+                },
+                "set": {
+                        "type": "object",
+                        "description": "Column name -> new value"
+                },
+                "where_clause": {
+                        "type": "string",
+                        "description": "Raw SQL predicate selecting the rows to update (required)"
+                }
+        }
+},
+        pathFieldRenames: undefined,
+        queryFieldRenames: undefined,
+        bodyFieldRenames: undefined
+    },
     {
         name: "createVariable",
         description: "create variable",