Agentic 基础原语

本页按你通常会遇到的顺序，介绍 Ageniti 提供的 5 个核心原语。

1. Action Contract

每个可调用能力都是一个 action：类型化的输入输出对，加上声明式的副作用策略。Contract 是所有 surface 的事实来源。

import { defineAction, s } from "@ageniti/core";
 
export const createTask = defineAction({
  name: "create_task",
  description: "Create a task in the user's inbox.",
  sideEffects: "write",
  idempotency: "conditional",
  input: s.object({
    title: s.string().min(1),
    priority: s.enum(["low", "high"]).default("low"),
  }),
  output: s.object({ id: s.string(), title: s.string() }),
  async run({ title, priority }, ctx) {
    ctx.logger.info("Creating task", { title });
    const task = await ctx.services.tasks.create({ title, priority });
    return { id: task.id, title: task.title };
  },
});

必填字段是 name、description、run。与内置 CLI 命令冲突的保留名称（actions、mcp、dev 等）会在定义时直接报错。

2. 用你已有的 Schema

defineAction 接受任何实现 Standard Schema v1 的 schema，或任何 Zod 风格的 schema（带 .safeParse / .parse）：

import { z } from "zod";
import { defineAction } from "@ageniti/core";
 
export const search = defineAction({
  name: "search_tasks",
  description: "Search for tasks matching a query.",
  input: z.object({
    query: z.string(),
    limit: z.number().int().optional(),
  }),
  output: z.object({
    results: z.array(z.object({ id: z.string(), title: z.string() })),
  }),
  async run({ query, limit }) {
    return { results: await tasks.search(query, limit ?? 20) };
  },
});

外部 schema 通过 duck typing 识别。MCP 工具定义和 OpenAI tool spec 所需的 JSON Schema 通过 best-effort introspection 生成（覆盖 ZodObject / ZodArray / ZodUnion / ZodOptional / ZodDefault 等常见类型）。

如果你的 schema 比较特殊，可以手动覆盖 JSON Schema：

import { wrapSchema } from "@ageniti/core/schema-adapter";
 
const wrapped = wrapSchema(myCustomSchema, {
  jsonSchema: { type: "object", properties: { … } },
});

3. 批量包装已有函数

针对 Next.js Server Actions、tRPC procedures、普通的 handler record：

import { actionsFromHandlers, s } from "@ageniti/core";
import * as handlers from "@/app/actions/tasks"; // 你已有的函数
 
export const actions = actionsFromHandlers(handlers, {
  createTask: {
    description: "Create a task.",
    input: s.object({ title: s.string() }),
    sideEffects: "write",
  },
  searchTasks: {
    description: "Search tasks.",
    input: s.object({ query: s.string() }),
  },
});

或者用 defineActions 获得完全控制权（值可以是 config 对象，也可以是只读无入参的简化形式）：

import { defineActions, s } from "@ageniti/core";
 
export const actions = defineActions({
  createTask: {
    description: "Create a task.",
    input: s.object({ title: s.string() }),
    run: async ({ title }) => tasks.create({ title }),
  },
  ping: () => ({ ok: true, time: Date.now() }),
});

camelCase 的 key 自动转成 snake_case 作为 action name。

4. 流式事件

每个 action 通过 runtime 执行，runtime 在执行过程中实时发出 live events。任何消费者（UI / agent / log shipper）都能通过 runtime.stream() 订阅：

const events = runtime.stream("create_task", { title: "Ship v1" });
 
for await (const event of events) {
  if (event.type === "log") console.log(event.level, event.message);
  if (event.type === "progress") updateProgressBar(event.percent);
  if (event.type === "artifact") attachToUi(event.artifact);
  if (event.type === "result") finalize(event.envelope);
}

事件来自 run() 函数里调用的 ctx.logger.* / ctx.progress.report() / ctx.artifacts.add()。CLI 的 --ndjson 模式和 React useAction hook 都是建在这个原语之上。

5. 类型化 Client + Codegen

类型化 client 包装任意 runtime 或远程 @ageniti HTTP server。成功时返回校验后的 data，失败时抛 AgenitiClientError。远端 HTTP 模式可以传 metadata、confirm 和 idempotencyKey，但可信的 user / auth 需要在服务端解析：

import { createClient } from "@ageniti/core/client";
 
// 进程内
const client = createClient({ runtime });
const task = await client.create_task({ title: "Hello" });
 
// 或者连远程 @ageniti HTTP server
const remote = createClient({ url: "https://api.example.com" });
const tasks = await remote.search_tasks({ query: "today" });

为消费方生成 .d.ts 让 IDE 有完整自动补全：

import { generateClientTypes } from "@ageniti/core/client-gen";
import { writeFile } from "node:fs/promises";
 
await writeFile(".ageniti/client.d.ts", generateClientTypes(actions, {
  interfaceName: "TasksClient",
}));

React Hook（建在流式事件之上）

React UI 用 useAction hook 订阅 runtime.stream，调用过程中 logs / artifacts / progress 实时更新：

import { useAction } from "@ageniti/core/react-hooks";
 
function CreateTaskButton() {
  const { invoke, status, data, error, logs, progress, cancel } =
    useAction(createTask, { runtime });
 
  return (
    <>
      <button onClick={() => invoke({ title: "Hello" })}
              disabled={status === "loading"}>
        {status === "loading" ? `${progress?.percent ?? 0}%` : "Create"}
      </button>
      {status === "loading" && <button onClick={cancel}>Cancel</button>}
      {status === "success" && <p>Created task {data.id}</p>}
      {status === "error" && <p>Error: {error.message}</p>}
    </>
  );
}

状态机：idle → loading → (success | error | cancelled)。组件 unmount 自动 abort，新的 invoke 调用会取消之前的。

测试工具

import {
  createTestRuntime, expectOk, expectError, collectStream,
} from "@ageniti/core/test-utils";
 
test("create_task happy path", async () => {
  const t = createTestRuntime([createTask], { services: { tasks: stubTasks } });
  const env = await t.invoke("create_task", { title: "Hello" });
  const data = expectOk(env);
  expect(data.title).toBe("Hello");
});

框架无关 —— 兼容 node:test、vitest、jest。createTestRuntime 默认 surface: "json" 并跳过 confirmation gate，方便测试 destructive action。

上面这 5 个原语是基础。其他每一个 surface —— CLI 生成、MCP server、OpenAI tool spec、AI SDK tools、HTTP server —— 都是同一份 action contract 和同一个流式 runtime 之上的薄适配层。你在它之上构建的任何东西（agent、planner、workflow、自定义 UI）读的是同一份 contract，订阅的是同一组事件。