feat(mcp): aggregate_records 工具 — GROUP BY 聚合走 engine 读路径（RLS/D10/FLS 全继承）

.changeset/mcp-aggregate-records.md

@@ -0,0 +1,29 @@
+---
+'@objectstack/mcp': minor
+'@objectstack/runtime': minor
+'@objectstack/plugin-security': patch
+---
+
+feat(mcp): `aggregate_records` tool — GROUP BY aggregation over the engine read path
+
+New MCP tool `aggregate_records` (count/sum/avg/min/max/count_distinct, optional
+groupBy incl. date bucketing, where filter, IANA timezone) in the `data:read`
+family. Execution routes through the ObjectQL ENGINE (`callData('aggregate')`
+deliberately never uses the raw per-env driver), so RLS/tenant scoping and the
+D10 delegator intersection apply exactly as on find.
+
+Security hardening shipped with it:
+
+- plugin-security: new FLS aggregate-INPUT gate — result masking never runs for
+  `aggregate` (output rows carry only aliases), so any groupBy / aggregation
+  reference to an FLS-unreadable field is now rejected fail-closed with the
+  offending field names (mirrors the FLS write gate).
+- runtime: `aggregate` maps to the `list` ApiMethod in the object exposure gate
+  (an object whose `apiMethods` whitelist excludes `list` cannot leak row
+  statistics through GROUP BY), and the aggregate action requires at least one
+  aggregation (the engine's in-memory path would otherwise degrade to raw rows
+  that the FLS masker does not cover).
+
+The bridge seam is optional: a runtime that does not implement
+`McpDataBridge.aggregate` simply does not register the tool (graceful
+degradation, same contract as the action tools).

content/docs/ai/natural-language-queries.mdx

@@ -1,6 +1,6 @@
 ---
 title: Natural Language Queries
-description: How agents query live data through the open MCP data tools (query_records, get_record) under RLS
+description: How agents query live data through the open MCP data tools (query_records, get_record, aggregate_records) under RLS
 ---
 
 # Natural Language Queries
@@ -15,19 +15,23 @@ back this are part of the open framework — no ObjectOS runtime and no
 cloud Studio are required.
 
 Your AI queries your data through the open MCP **data tools** — `query_records`
-(filter, field selection, sort, pagination) and `get_record`, plus the
-discovery tools `list_objects` and `describe_object` — which the model calls
-with structured arguments. These run as ordinary [ObjectQL](/docs/protocol/objectql) queries over
+(filter, field selection, sort, pagination), `get_record`, and `aggregate_records`
+(group-by count/sum/avg/min/max/count_distinct, with optional date bucketing),
+plus the discovery tools `list_objects` and `describe_object` — which the model
+calls with structured arguments. These run as ordinary [ObjectQL](/docs/protocol/objectql) queries over
 your objects (ObjectStack uses ObjectQL, not SOQL), and they execute under the
 caller's `ExecutionContext`, so row-level security applies exactly as it does
 for the REST API.
 
 <Callout type="info">
-**Aggregation is a cloud data tool.** The open MCP surface reads and filters
-rows; it does not ship a server-side `aggregate_data` (group-by / roll-up)
-tool — `query_records` has no aggregation arguments. A BYO agent aggregates by
-fetching the rows it's allowed to see and summing client-side; the built-in
-`aggregate_data` tool is part of the cloud in-product chat runtime (below).
+**Server-side aggregation is open too.** `aggregate_records` runs a single-object
+GROUP BY (count/sum/avg/min/max/count_distinct, optional date bucketing and
+filter) through the same ObjectQL engine read path — so totals and breakdowns
+come back in one call under row-level **and** field-level security, instead of
+the agent paging every visible row and summing client-side. It does not join
+across objects or read datasets/cubes: the cross-object, dataset-level
+`aggregate_data` roll-up tool remains part of the cloud in-product chat runtime
+(below).
 </Callout>
 
 The open MCP server plugin exposes these tools automatically — it bridges the
@@ -45,7 +49,7 @@ await kernel.bootstrap();
 
 Point any MCP client at the server and ask questions in natural language: the
 model discovers `list_objects` / `describe_object` / `query_records` /
-`get_record` and calls them under RLS as the authenticated caller.
+`get_record` / `aggregate_records` and calls them under RLS as the authenticated caller.
 
 There is no separate natural-language-to-query metadata type to author — the
 model is prompted with the available objects and translates the user's question
@@ -55,7 +59,7 @@ into `query_records` calls at runtime.
 **ObjectOS — bundled in-product chat.** ObjectOS ships an in-UI
 AI runtime — the `ask` data-query assistant
 and the `/api/v1/ai/*` chat endpoints — that registers these same data tools,
-plus the group-by/roll-up `aggregate_data` tool, into its own chat loop. That
+plus the dataset-level (cross-object) `aggregate_data` roll-up tool, into its own chat loop. That
 runtime is not part of the open framework and is documented separately in the
 [ObjectOS docs](https://docs.objectos.app/docs/ai); this page covers the open path
 only. On the open-source framework, use

packages/mcp/src/mcp-http-tools.scopes.test.ts

@@ -22,7 +22,7 @@ import {
 import { MCPServerRuntime } from './mcp-server-runtime.js';
 import type { McpActionBridge, McpDataBridge } from './mcp-http-tools.js';
 
-const READ_TOOLS = ['list_objects', 'describe_object', 'query_records', 'get_record'];
+const READ_TOOLS = ['list_objects', 'describe_object', 'query_records', 'get_record', 'aggregate_records'];
 const WRITE_TOOLS = ['create_record', 'update_record', 'delete_record'];
 const ACTION_TOOLS = ['list_actions', 'run_action'];
 
@@ -34,6 +34,7 @@ function makeBridge(): McpDataBridge & McpActionBridge & { calls: any[] } {
     async describeObject(name: string) { calls.push(['describeObject', name]); return { name }; },
     async query(object: string, opts: any) { calls.push(['query', object, opts]); return { object, records: [] }; },
     async get(object: string, id: string) { calls.push(['get', object, id]); return { id }; },
+    async aggregate(object: string, opts: any) { calls.push(['aggregate', object, opts]); return []; },
     async create(object: string, data: any) { calls.push(['create', object, data]); return { object, id: 'n1' }; },
     async update(object: string, id: string, data: any) { calls.push(['update', object, id, data]); return { object, id }; },
     async remove(object: string, id: string) { calls.push(['remove', object, id]); return { object, id, deleted: true }; },

packages/mcp/src/mcp-http-tools.ts

@@ -56,6 +56,22 @@ export interface McpDataBridge {
   create(object: string, data: Record<string, unknown>): Promise<unknown>;
   update(object: string, id: string, data: Record<string, unknown>): Promise<unknown>;
   remove(object: string, id: string): Promise<unknown>;
+  /**
+   * GROUP BY aggregation through the ObjectQL engine's read path (RLS + the
+   * FLS aggregate-input gate). OPTIONAL: a runtime that cannot route
+   * aggregation through the engine simply omits it and the
+   * `aggregate_records` tool is not registered (graceful degradation, same
+   * contract as {@link McpActionBridge}).
+   */
+  aggregate?(
+    object: string,
+    opts: {
+      where?: Record<string, unknown>;
+      groupBy?: Array<string | { field: string; dateGranularity?: string; alias?: string }>;
+      aggregations: Array<{ function: string; field?: string; alias: string; distinct?: boolean }>;
+      timezone?: string;
+    },
+  ): Promise<unknown[]>;
 }
 
 export interface RegisterObjectToolsOptions {
@@ -282,6 +298,82 @@ export function registerObjectTools(
       },
     );
 
+    if (typeof bridge.aggregate === 'function') {
+      const aggregateFn = bridge.aggregate.bind(bridge);
+      server.registerTool(
+        'aggregate_records',
+        {
+          description:
+            'Aggregate records with GROUP BY: count/sum/avg/min/max/count_distinct over an object, ' +
+            'optionally grouped by fields (dates can be bucketed by day/week/month/quarter/year). ' +
+            'Use this instead of paging query_records when a question needs totals or breakdowns. ' +
+            'Runs under the caller\'s permissions, row-level security and field-level security.',
+          inputSchema: {
+            objectName: z.string().describe('The object/table name'),
+            aggregations: z
+              .array(
+                z.object({
+                  function: z
+                    .enum(['count', 'sum', 'avg', 'min', 'max', 'count_distinct'])
+                    .describe('Aggregation function'),
+                  field: z.string().optional().describe('Field to aggregate (omit for count(*))'),
+                  alias: z.string().describe('Result column name'),
+                }),
+              )
+              .min(1)
+              .describe('Metrics to compute, e.g. [{"function":"sum","field":"amount","alias":"total"}]'),
+            groupBy: z
+              .array(
+                z.union([
+                  z.string(),
+                  z.object({
+                    field: z.string().describe('Field to group by'),
+                    dateGranularity: z
+                      .enum(['day', 'week', 'month', 'quarter', 'year'])
+                      .optional()
+                      .describe('Bucket a date field into uniform periods'),
+                    alias: z.string().optional().describe('Alias for the projected group value'),
+                  }),
+                ]),
+              )
+              .optional()
+              .describe('Grouping fields; omit for a single overall row'),
+            where: z
+              .record(z.string(), z.unknown())
+              .optional()
+              .describe('Filter conditions applied before aggregation, e.g. {"status":"open"}'),
+            timezone: z
+              .string()
+              .optional()
+              .describe('IANA timezone for date bucketing (defaults to UTC)'),
+          },
+          annotations: { readOnlyHint: true, destructiveHint: false, openWorldHint: false },
+        },
+        async ({ objectName, aggregations, groupBy, where, timezone }) => {
+          const bad = guard(objectName);
+          if (bad) return errorResult(bad);
+          try {
+            const rows = (await aggregateFn(objectName, {
+              where,
+              groupBy,
+              aggregations,
+              timezone,
+            })) ?? [];
+            // Group count is unbounded (a high-cardinality groupBy can return
+            // thousands of rows) — cap the tool output like query_records does.
+            const truncated = rows.length > maxLimit;
+            return textResult({
+              rows: truncated ? rows.slice(0, maxLimit) : rows,
+              totalGroups: rows.length,
+              ...(truncated ? { truncated: true } : {}),
+            });
+          } catch (err) {
+            return errorResult(messageOf(err));
+          }
+        },
+      );
+    }
+
     server.registerTool(
       'get_record',
       {

packages/mcp/src/mcp-server-runtime.http.test.ts

@@ -223,3 +223,164 @@ describe('MCPServerRuntime.handleHttpRequest (Streamable HTTP)', () => {
     expect(json.result.content[0].text).toContain('RLS: not permitted');
   });
 });
+
+describe('aggregate_records (GROUP BY aggregation tool)', () => {
+  let runtime: MCPServerRuntime;
+
+  /** Bridge WITH the optional aggregate seam implemented. */
+  function makeAggBridge(rows: unknown[] = [{ status: 'open', n: 2 }]) {
+    const base = makeBridge();
+    const calls = base.calls;
+    return {
+      ...base,
+      calls,
+      async aggregate(object: string, opts: any) {
+        calls.push(['aggregate', object, opts]);
+        return rows;
+      },
+    };
+  }
+
+  beforeEach(() => {
+    runtime = new MCPServerRuntime({ name: 'objectstack-test', version: '9.9.9' });
+  });
+
+  it('registers only when the bridge implements aggregate (graceful degradation)', async () => {
+    const without = await call(runtime, { jsonrpc: '2.0', id: 1, method: 'tools/list' }, makeBridge());
+    expect(without.json.result.tools.map((t: any) => t.name)).not.toContain('aggregate_records');
+
+    const runtime2 = new MCPServerRuntime({ name: 'objectstack-test', version: '9.9.9' });
+    const withAgg = await call(runtime2, { jsonrpc: '2.0', id: 1, method: 'tools/list' }, makeAggBridge());
+    const byName = Object.fromEntries(withAgg.json.result.tools.map((t: any) => [t.name, t]));
+    expect(byName.aggregate_records).toBeDefined();
+    expect(byName.aggregate_records.annotations.readOnlyHint).toBe(true);
+  });
+
+  it('delegates groupBy/aggregations/where to the bridge and returns rows', async () => {
+    const bridge = makeAggBridge([{ status: 'open', n: 2 }, { status: 'done', n: 5 }]);
+    const { json } = await call(
+      runtime,
+      {
+        jsonrpc: '2.0',
+        id: 2,
+        method: 'tools/call',
+        params: {
+          name: 'aggregate_records',
+          arguments: {
+            objectName: 'task',
+            groupBy: ['status', { field: 'created_at', dateGranularity: 'month' }],
+            aggregations: [{ function: 'count', alias: 'n' }],
+            where: { active: true },
+          },
+        },
+      },
+      bridge,
+    );
+    expect(json.result.isError).toBeFalsy();
+    const payload = JSON.parse(json.result.content[0].text);
+    expect(payload.rows).toHaveLength(2);
+    expect(payload.totalGroups).toBe(2);
+    expect(payload.truncated).toBeUndefined();
+    const aggCall = bridge.calls.find((c: any[]) => c[0] === 'aggregate');
+    expect(aggCall[1]).toBe('task');
+    expect(aggCall[2].groupBy).toEqual(['status', { field: 'created_at', dateGranularity: 'month' }]);
+    expect(aggCall[2].aggregations).toEqual([{ function: 'count', alias: 'n' }]);
+    expect(aggCall[2].where).toEqual({ active: true });
+  });
+
+  it('rejects system objects fail-closed without consulting the bridge', async () => {
+    const bridge = makeAggBridge();
+    const { json } = await call(
+      runtime,
+      {
+        jsonrpc: '2.0',
+        id: 3,
+        method: 'tools/call',
+        params: {
+          name: 'aggregate_records',
+          arguments: {
+            objectName: 'sys_user',
+            aggregations: [{ function: 'count', alias: 'n' }],
+          },
+        },
+      },
+      bridge,
+    );
+    expect(json.result.isError).toBe(true);
+    expect(json.result.content[0].text).toMatch(/system object/i);
+    expect(bridge.calls.find((c: any[]) => c[0] === 'aggregate')).toBeUndefined();
+  });
+
+  it('rejects an empty aggregations array (schema min(1))', async () => {
+    const bridge = makeAggBridge();
+    const { json } = await call(
+      runtime,
+      {
+        jsonrpc: '2.0',
+        id: 4,
+        method: 'tools/call',
+        params: {
+          name: 'aggregate_records',
+          arguments: { objectName: 'task', aggregations: [] },
+        },
+      },
+      bridge,
+    );
+    // Zod input validation fails before the handler runs — surfaced as a
+    // tool error or JSON-RPC error depending on SDK version; both are fine,
+    // the invariant is that the bridge is never reached.
+    expect(json.result?.isError === true || json.error != null).toBe(true);
+    expect(bridge.calls.find((c: any[]) => c[0] === 'aggregate')).toBeUndefined();
+  });
+
+  it('caps the returned group rows at the query limit with a truncated flag', async () => {
+    const manyGroups = Array.from({ length: 60 }, (_, i) => ({ k: `g${i}`, n: i }));
+    const bridge = makeAggBridge(manyGroups);
+    const { json } = await call(
+      runtime,
+      {
+        jsonrpc: '2.0',
+        id: 5,
+        method: 'tools/call',
+        params: {
+          name: 'aggregate_records',
+          arguments: {
+            objectName: 'task',
+            groupBy: ['k'],
+            aggregations: [{ function: 'count', alias: 'n' }],
+          },
+        },
+      },
+      bridge,
+    );
+    const payload = JSON.parse(json.result.content[0].text);
+    expect(payload.rows).toHaveLength(50); // DEFAULT_MAX_LIMIT
+    expect(payload.totalGroups).toBe(60);
+    expect(payload.truncated).toBe(true);
+  });
+
+  it('surfaces bridge errors (e.g. FLS denial) as tool errors', async () => {
+    const bridge = makeAggBridge();
+    bridge.aggregate = async () => {
+      throw new Error('[Security] Field read denied: not permitted to aggregate [salary]');
+    };
+    const { json } = await call(
+      runtime,
+      {
+        jsonrpc: '2.0',
+        id: 6,
+        method: 'tools/call',
+        params: {
+          name: 'aggregate_records',
+          arguments: {
+            objectName: 'employee',
+            aggregations: [{ function: 'sum', field: 'salary', alias: 'total' }],
+          },
+        },
+      },
+      bridge,
+    );
+    expect(json.result.isError).toBe(true);
+    expect(json.result.content[0].text).toContain('Field read denied');
+  });
+});

packages/mcp/src/mcp-server-runtime.ts

@@ -51,6 +51,7 @@ const READ_ONLY_TOOLS = new Set([
   'describe_object',
   'query_records',
   'get_record',
+  'aggregate_records',
   'aggregate_data',
 ]);