Mastra

查看 Markdown

Mastra 是一个用于 Agent、tool 和 workflow 的 TypeScript framework。Dex adapter 把 Mastra Agent 放进 durable session 中运行。它支持 TypeScript。

何时使用

Agent 已经用 Mastra 编写时，使用它。保留 Agent、它的 tool 和 model router，由 Dex 管理 session。

Session 如何运行

Agent 的每个 tool，包括 read tool，都作为 client tool 交给 Mastra。因此 Mastra 在模型调用之后停下，带回 tool call。每个 RunModel Step 调用一次模型。

每个 tool call 在自己的 Step 中并行运行 Agent 原本的 tool 一次：read 或 idempotent write 用 CallTool，其他写操作用 CallSingleAttemptTool。需要审批的调用先在 AwaitToolApproval 中等待。下一个 RunModel 加入记录下来的结果，然后再次调用模型。

查询订单的 turn 依次运行 RunModel、CallTool、ResumeTurn 和 RunModel。

安装

把 adapter 与它的 framework 一起安装。每个 adapter 都会带上 Dex Agents core。

npm i @superdurable/dex-agents-mastra @superdurable/dex-agents @superdurable/dex @mastra/core zod

例子: agents/typescript/mastra/README.md

Example

这个 example 就是每个 framework 页面共用的订单客服 Agent。

Flow type 是 MastraOrderSupportAgent（TypeScript）。完整 example 位于 examples/typescript/src/products/agent-harnesses。

Agent 与 tool

Tool 是 Mastra Agent 上的 Mastra tool。Tool policy 指定每个 tool。

mastra-order-support-flow.tsTypeScript
1234567891011121314151617181920212223242526272829303132
export function createMastraOrderSupportFlow(
  orderDesk: OrderDesk,
  model: MastraModelConfig,
  slack: Connection | undefined,
): AgentSessionFlow {
  const lookUpOrder = createTool({
    id: "look_up_order",
    description: "Look up one order by its order ID and return its items and status.",
    inputSchema: z.object({ order_id: z.string() }),
    execute: async ({ order_id }) => orderDesk.describeOrder(order_id),
  });
  const cancelOrder = createTool({
    id: "cancel_order",
    description: "Cancel one order that has not shipped yet.",
    inputSchema: z.object({ order_id: z.string(), reason: z.string() }),
    execute: async ({ order_id, reason }) => orderDesk.cancelOrder(order_id, reason),
  });
  const agent = new Agent({
    id: "order-support",
    name: "OrderSupport",
    instructions: orderSupportInstructions,
    model,
    tools: { look_up_order: lookUpOrder, cancel_order: cancelOrder },
  });
  return createMastraAgentSessionFlow(agent, mastraOrderSupportFlowType, {
    toolPolicies: {
      look_up_order: { effect: "read" },
      cancel_order: { effect: "non_idempotent_write" },
    },
    toolSources: createSlackConnectorToolSources(slack),
  });
}

例子: examples/typescript/src/products/agent-harnesses/mastra/mastra-order-support-flow.ts

Worker

一个 Worker 为 example 中所有 harness 的 session Flow 提供服务。每个启动 session 或调用其 RPC 的进程，都要构建同一个 session Flow。

registry.tsTypeScript
1234567
// One Slack connection for the agent examples; without it they offer no Slack tools.
const slack = await openSlackConnectionFromProject();
// ...
// One order desk for the four agent harness examples; each harness registers its own session Flow type.
const orderDesk = new OrderDesk();
// ...
const mastraOrderSupportFlow = createMastraOrderSupportFlow(orderDesk, loadMastraOrderSupportModel(), slack);

例子: examples/typescript/src/registry.ts

Session route

每个 harness 都提供这同一组 route。这些 route 和 session token 的说明见 Session route。

order-support-routes.tsTypeScript
1234567891011121314151617181920
  router.post("/sessions", async (request, response) => {
    const initialMessage = optionalString(requireBody(request), "message");
    // The server picks the session ID, so a caller cannot name another session.
    const sessionId = `order-support-${randomUUID()}`;
    await startAgentSession(client, flow, sessionId, { initialMessage }, { timeoutMs: 30 * DAY_MS });
    response.json({ session_id: sessionId, session_token: issueSessionToken(sessionId) });
  });
  router.get("/sessions/:id/snapshot", async (request, response) => {
    response.json(convertAgentSnapshotToJson(await client.invokeRPC(flow.getAgentSnapshot, sessionIdOf(request))));
  });
  router.post("/sessions/:id/messages", async (request, response) => {
    const sessionId = sessionIdOf(request);
    const message = { text: requireString(requireBody(request), "text") };
    sendAccepted(response, await client.invokeRPC(flow.sendUserMessage, sessionId, message));
  });
  router.post("/sessions/:id/interrupts/:interruptId/approve", async (request, response) => {
    const sessionId = sessionIdOf(request);
    const approval = { interruptId: String(request.params.interruptId), approved: true, rejectionMessage: null };
    sendAccepted(response, await client.invokeRPC(flow.resolveInterrupt, sessionId, approval));
  });

例子: examples/typescript/src/products/agent-harnesses/order-support/order-support-routes.ts

在本地运行

启动 Dex 与 example

在第一个终端中，先渲染 session graph，再启动 Dex。Dex Web 启动时读取这个 graph，用它显示 session 和审批控件。

$$$$$
cd examples/typescript
npm install
mkdir -p definitions
npx dex-agents-session-graph --flow-type MastraOrderSupportAgent \
  --permission order-support.operate --out definitions/mastra.json
dexcli dev --flow-rendering-dir ./definitions

例子: examples/typescript/src/products/agent-harnesses/README.md

dexcli dev 会一直在前台运行。在第二个终端中启动 example：

$$
cd examples/typescript
OPENAI_API_KEY=... MASTRA_TELEMETRY_DISABLED=1 npm start

例子: examples/typescript/src/products/agent-harnesses/README.md

应用监听 127.0.0.1:8080。Dex Web 位于 127.0.0.1:8802。

这个 example 没有 demo model。它的 model router 从 Worker 的环境中读取 provider key，例如 OPENAI_API_KEY。

变量	默认值	含义
DEX_MASTRA_MODEL	openai/gpt-5-mini	Mastra model router ID。
OPENAI_API_KEY	无；使用默认模型时必填	Router 使用的 provider key。
MASTRA_TELEMETRY_DISABLED	未设置	设为 1 可关闭 Mastra 的 telemetry。
DEX_AGENT_SESSION_KEY	随机 key	签发 session token 的 key。

调用 session

启动一个 session，让 Agent 取消订单 1002，然后批准这次调用：

$$$$$$
base=http://127.0.0.1:8080/products/agent-harnesses/mastra
session=$(curl -s -X POST "$base/sessions" -H 'Content-Type: application/json' \
  -d '{"message": "Please cancel order 1002"}')
id=$(echo "$session" | jq -r .session_id)
token=$(echo "$session" | jq -r .session_token)
until interrupt=$(curl -s "$base/sessions/$id/snapshot" -H "X-Agent-Session-Token: $token" \
  | jq -re '.pending_interrupts[0].interrupt_id'); do sleep 1; done
curl -s -X POST "$base/sessions/$id/interrupts/$interrupt/approve" -H "X-Agent-Session-Token: $token"

例子: examples/typescript/src/products/agent-harnesses/README.md

循环会反复读取 snapshot，直到 Agent 请求运行 cancel_order。这次调用先等待审批，批准后只运行一次。再读一次 snapshot 即可看到 Agent 的回答。

在 Dex Web 中审批

也可以不执行最后一条命令，改在 Dex Web 中审批，步骤见在 Dex Web 中审批。

连接 Connector

Agent 可以把 Connector 的 operation 当作 tool 调用，见把 Connector 用作 tool。这个 example 加入了 Slack tool slack_post_thread_reply，用来在 Slack thread 中回复。

提供 tool

安装 @superdurable/connectors-sdk 以及 Connector 包，例如 @superdurable/connector-slack。安装命令见 Connector tools。

Worker 从项目配置中打开 Slack connection support-slack。connectorTool 把 operation 变成 tool，由 ConnectorToolSource 持有。没有这个 connection 时，agent 不会得到 Slack tool。

slack-connector-tools.tsTypeScript
1234567891011121314151617181920212223242526
export async function openSlackConnectionFromProject(
  environment: NodeJS.ProcessEnv = process.env,
): Promise<Connection | undefined> {
  // ...
  const project = await loadFromEnvironment(environment);
  // ...
  return newProjectConnection(project, slackConnectionName);
}

// ...

export function createSlackConnectorToolSources(slack: Connection | undefined): ToolSource[] {
  if (slack === undefined) {
    return [];
  }
  return [
    new ConnectorToolSource([
      connectorTool(slack.operations.postThreadReply(), slack.reference, {
        manifest,
        inputValidator: PostThreadReplyInputSchema,
        // The manifest does not ask for approval; this agent asks before every post.
        requiresApproval: true,
      }),
    ]),
  ];
}

例子: examples/typescript/src/shared/slack-connector-tools.ts

Worker 在启动时打开一次 connection，每个 agent 都使用它：

registry.tsTypeScript
12
// One Slack connection for the agent examples; without it they offer no Slack tools.
const slack = await openSlackConnectionFromProject();

例子: examples/typescript/src/registry.ts

Agent 把 tool source 和自己的 tool 一起传入：

mastra-order-support-flow.tsTypeScript
1234
  return createMastraAgentSessionFlow(agent, mastraOrderSupportFlowType, {
    // ...
    toolSources: createSlackConnectorToolSources(slack),
  });

例子: examples/typescript/src/products/agent-harnesses/mastra/mastra-order-support-flow.ts

调用如何运行

每次调用 slack_post_thread_reply 都要先经人批准，然后在自己的 CallSingleAttemptTool Step 中只运行一次。原因见 Slack Connector tool。

Adapter 把 Connector tool 作为没有 execute 的 client tool 传给 Mastra。调用从不在模型调用内运行。Mastra 的审批设置对它不起作用；审批由 source 决定。使用 tool source 时，toolPolicies 条目从不会让一个 tool 变成由 Mastra 自己运行的 tool；只有 mastraRunToolNames 会。

配置 connection

Worker 从哪里找到 connection support-slack、Slack app 需要什么，见 Connection。

试用时，先设置 DEX_CONNECTOR_CONFIG_FILE 再启动 example。然后让 agent 在 Slack thread 中回复，并给出 channel ID 和 thread timestamp：

Reply in Slack thread C0123456789 1710000000.000100: Order 1002 ships on Thursday.

这次发帖和 cancel_order 一样，会先等待审批。

限制

Mastra 自己运行的 tool，例如 sub-agent、workflow 和 Memory tool，无法在 Step 中运行。只有当 toolPolicies 把它标记为无需审批的 read 时，adapter 才会把它提供给模型。这时它在 RunModel 内运行，并在这个 Step 重跑时重复执行。
不使用 Mastra Memory。不要在 Agent 的默认设置或 scope 中选择 Memory thread。
beforeToolCall 与 afterToolCall hook 只对 Mastra 自己运行的 tool 生效。
Tool result 以文本形式交给模型。不会应用 toModelOutput。
调用 suspend() 的 tool 会失败。请改用 session 的审批。
Tool 名称最多 63 个字符，只能包含字母、数字、_ 和 -，并且不能以数字或 - 开头。

Adapter README 列出了全部限制。Server 的限制参见 Limits for long sessions。

何时使用​

Session 如何运行​

安装​

Example​

Agent 与 tool​

Worker​

Session route​

在本地运行​

启动 Dex 与 example​

调用 session​

在 Dex Web 中审批​

连接 Connector​

提供 tool​

调用如何运行​

配置 connection​

限制​

相关内容​

何时使用

Session 如何运行

安装

Example

Agent 与 tool

Worker

Session route

在本地运行

启动 Dex 与 example

调用 session

在 Dex Web 中审批

连接 Connector

提供 tool

调用如何运行

配置 connection

限制

相关内容