Agent Controller

AgentController 是交互式 agent 应用的共享 runtime 宿主。它协调模式、模型、存储、workspaces、工具审批、subagents 和渠道。每个用户或活动任务都通过一个隔离的 Session 工作。

Mastra Code 和 Mastra Factory 是 AgentController 的旗舰实现。它们是具有多模型支持、持久对话和“先计划后执行”工作流的编程 agents。阅读构建编程 agent获取分步 TUI 指南。

当你托管的 agent 在代码库中工作时，请使用 createCodingAgent() 构建它，而不是 new Agent()。它返回一个标准 Agent，该 agent 已经具备 workspace、任务跟踪以及针对瞬时模型错误的重试，这些正是 Mastra Code 运行的默认配置。

何时使用 Agent Controller

当你的应用需要以下能力时，使用 Agent Controller：

• 多个 agent 模式共享同一个对话 thread（例如 plan → build → review）
• 在 UI 和 agent 循环之间有一个控制层（模型切换、状态持久化、thread 管理）
• 工具审批流程和权限策略，用于 human-in-the-loop 门控
• Subagent 编排，用受限工具委派聚焦的子任务
• 跨重启的持久 threads 和选定的 thread 设置，同时为每个 Session 提供隔离的实时状态

你可以在 Agent 类之上自行组装所有这些功能，该类暴露了完整的 agent 循环、工具和记忆。AgentController 为持续会话提供有主见的默认值，在这种会话中 agent 扮演协作者而非一次性端点。需要完全控制或请求-响应调用时，直接使用 Agent 类。选择 AgentController 可以在无需自建会话 runtime 的情况下获得持续会话。

快速开始

创建底层的 Agent、存储和 Workspace。调用一次 controller.init()，然后使用 controller.createSession() 创建 Session。使用 session.subscribe() 订阅，并使用 session.sendMessage() 发送工作：

src/mastra/agent-controller.ts

import { Agent } from '@mastra/core/agent'
import { AgentController } from '@mastra/core/agent-controller'
import { LocalFilesystem, Workspace } from '@mastra/core/workspace'
import { LibSQLStore } from '@mastra/libsql'

const agent = new Agent({
  id: 'assistant',
  name: 'Assistant',
  instructions: 'Help the user plan and complete tasks.',
  model: 'openai/gpt-5.6-sol',
})

const controller = new AgentController({
  id: 'assistant-controller',
  agent,
  storage: new LibSQLStore({
    id: 'agent-controller-storage',
    url: 'file:./mastra.db',
  }),
  workspace: new Workspace({
    id: 'assistant-workspace',
    filesystem: new LocalFilesystem({ basePath: './workspace' }),
  }),
  modes: [
    {
      id: 'plan',
      name: 'Plan',
      metadata: { default: true },
      instructions: 'Reason about the task before making changes.',
    },
    {
      id: 'build',
      name: 'Build',
      instructions: 'Implement the approved plan.',
    },
  ],
})

await controller.init()

const session = await controller.createSession({
  resourceId: 'user-123',
})

const unsubscribe = session.subscribe(event => {
  if (event.type === 'message_update' && event.event.type === 'text-delta') {
    process.stdout.write(event.event.delta)
  }
})

await session.sendMessage({ content: 'Plan a small TypeScript CLI.' })
unsubscribe()

每条消息都会发出一个带有初始消息的 message_start 事件、零个或多个带有紧凑增量的 message_update 事件，以及一个包含消息 ID 的 message_end 事件。当你需要重建完整消息时，按 ID 应用更新。

同一个 controller 可用于多个 Sessions。不要在 controller 上存储“当前 Session”，也不要通过 controller 级消息方法路由工作。

理解 runtime 模型

Controller、Session 和 thread 具有不同的生命周期：

• Controller：配置和 runtime 服务的共享宿主。初始化一次并重复使用。
• Session：一个用户、任务或并发工作范围的隔离实时 runtime。它拥有活动模式、模型、状态、事件总线、运行状态、授权和当前 thread 绑定。
• Thread：存储的对话，包含消息和 thread 设置。配置存储后，threads 可以在 controller 和进程重建后继续存在。

Session 是实时状态。任意的 session.state、权限授权、待处理审批和活动运行不会在进程重建后自动保留。Thread 消息和选定的 thread 设置（包括活动模式和当前模型）可以通过存储持久化，但切换模式不会更改模型。

Sessions 与 threads

createSession() 按 resourceId 和可选的 scope 执行 get-or-create：

const webSession = await controller.createSession({
  resourceId: 'user-123',
  scope: 'web',
})

const sameWebSession = await controller.createSession({
  resourceId: 'user-123',
  scope: 'web',
})

const workerSession = await controller.createSession({
  resourceId: 'user-123',
  scope: 'background-worker',
})

console.log(webSession === sameWebSession) // true
console.log(webSession === workerSession) // false

不同 scope 的 Sessions 具有独立的事件总线、运行循环、状态、模式和模型选择，以及当前 thread 绑定。它们存储的 threads 仍属于共享的 resourceId。

当宿主必须将 Session 绑定到确切 thread 时，传入 threadId。Controller 会切换到现有 thread，或在它不存在时使用该 ID 创建。当 createSession() 返回缓存的 Session 时，此行为同样适用：

const session = await controller.createSession({
  resourceId: 'user-123',
  scope: 'web',
  threadId: 'support-ticket-42',
})

使用 session.thread.create() 和 session.thread.switch() 在一个实时 Session 的多个对话之间切换。

转移 thread 所有权

每个新 thread 都会将其持久所有者存储在 metadata.ownerId 中，并将其原始创建者存储在 metadata.createdBy 中。两者都默认使用 Session 宿主的 ownerId。所有权转移时，createdBy 不会改变。

读取活动 thread 的所有者，并通过 session.thread 转移它：

const currentOwnerId = await session.thread.getOwner()

await session.thread.transferOwnership({
  toOwnerId: 'user-456',
})

转移在单个进程内是串行化的。需要持久化跨进程 compare-and-set 的宿主，应先在自己的存储中提交，然后调用 transferOwnership() 将所有者镜像到活动 thread。

每次运行在启动时都会对 thread 所有者做一次快照。工具和其他动态 runtime 代码可以从 AgentControllerRequestContext.threadOwnerId 读取该值。如果所有权在运行期间发生变化，活动运行会保留旧快照，下一次运行会收到新所有者。context.session.ownerId 仍然是进程本地 Session 宿主的身份。

从客户端列出已存储消息

使用 Agent Controller 客户端分页浏览 thread 的已存储消息。传入选项对象会返回消息及分页元数据：

import { MastraClient } from '@mastra/client-js'

const client = new MastraClient({ baseUrl: 'http://localhost:4111' })
const session = client.getAgentController('assistant-controller').session('user-123')

const result = await session.listMessages('support-ticket-42', {
  page: 0,
  perPage: 20,
  orderBy: { field: 'createdAt', direction: 'DESC' },
  filter: {
    dateRange: { start: new Date('2026-01-01') },
  },
})

console.log(result.messages)
console.log(result.total)
console.log(result.hasMore)

page 从零开始索引。省略 perPage 时使用存储默认值 40 条消息。使用 include 按 ID 请求某条消息及其相邻消息。limit 是 perPage 的已弃用别名；现有的分页调用方可以使用 { limit, page }，但当使用 orderBy、filter 或 include 时必须使用 perPage。现有的数字形式 session.listMessages(threadId, limit) 仍可用于以数组形式获取最新消息窗口，按从旧到新排序。

切换模式和模型

模式会更改共享底层 agent 使用的指令和工具，而不会替换 Session 或 thread。在 controller 上配置模式特定的工具和可见性：

const modes = [
  {
    id: 'plan',
    name: 'Plan',
    metadata: { default: true },
    instructions: 'Investigate the task and propose a plan.',
    additionalTools: { searchDocs },
    availableTools: ['searchDocs', 'submit_plan'],
    transitionsTo: 'build',
  },
  {
    id: 'build',
    name: 'Build',
    instructions: 'Implement the approved plan.',
  },
]

tools 和 additionalTools 是用于添加模式特定工具的互斥输入。当 controller 有共享底层 agent 时，任一种输入都会将这些工具叠加到 agent 的工具上。使用 availableTools 限制某个模式最终暴露的工具名称。权限拒绝仍然优先于此允许列表。

使用 session.mode.switch() 切换活动 Session。使用 session.mode.get() 或 session.mode.resolve() 读取活动模式：

await session.mode.switch({ modeId: 'build' })

console.log(session.mode.get()) // "build"
console.log(session.mode.resolve().instructions)

使用 session.model.switch() 独立切换模型，然后使用 session.model.get() 读取活动选择。该选择会作为 thread 的当前模型存储，并在 Session 切换模式时保持活动：

await session.model.switch('anthropic/claude-sonnet-4-6')

console.log(session.model.get())

选择 Observational Memory 模型

Observer 和 reflector 的模型选择是独立的。每个角色都可以使用显式模型，或使用由宿主根据活动主模型和会话状态解析的自动选择。

使用返回具体模型 ID 的同步解析器配置自动选择：

const controller = new AgentController({
  // ...
  omConfig: {
    observerModel: 'auto',
    reflectorModel: 'auto',
    defaultObserverModelId: 'openai/gpt-5-mini',
    defaultReflectorModelId: 'openai/gpt-5-mini',
    resolveAutoModelId: ({ currentModelId }) => currentModelId,
  },
})

session.om.observer.model() 返回 'auto' 或具体模型 ID。session.om.observer.modelId() 返回实际生效的具体模型。因此，宿主可以在不重写自动角色设置的情况下更改主模型。

将一个角色重置为自动选择，而不更改另一个角色：

await session.om.observer.switchModel({ modelId: 'auto' })

如果解析器返回 undefined 或抛出异常，该角色会使用其具体的 defaultObserverModelId 或 defaultReflectorModelId 作为回退。关于显式选择，请参阅 Session Observational Memory API。

管理 threads 和状态

使用 session.thread.list() 列出已存储的对话：

const thread = await session.thread.create({ title: 'Release planning' })
const threads = await session.thread.list()

await session.thread.switch({ threadId: thread.id })
console.log(threads.length)

使用 session.state 表示与 Session 关联的结构化实时状态。使用 session.state.get() 读取它，使用 session.state.set() 写入更新。当需要校验和默认值时，在 controller 上定义 stateSchema 和 initialState：

console.log(session.state.get())

await session.state.set({ activeProject: 'docs-site' })

session.state.get() 返回快照。set() 会校验更新并将其合并到 Session 状态中。除非你的宿主显式持久化并恢复它，否则请将此状态视为实时 Session 数据。

审批工具并恢复挂起

权限策略决定某个工具是被允许、拒绝，还是发送到 UI 等待审批。在 controller 上使用 toolCategoryResolver 将自定义工具映射到类别：

const controller = new AgentController({
  toolCategoryResolver: toolName => {
    if (toolName === 'delete_project') return 'execute'
    return null
  },
})

使用 session.permissions.setForCategory() 配置类别策略，使用 session.permissions.setForTool() 配置工具策略：

await session.permissions.setForCategory({
  category: 'execute',
  policy: 'ask',
})

await session.permissions.setForTool({
  toolName: 'delete_project',
  policy: 'deny',
})

当策略解析为 ask 时，订阅审批事件，并使用 session.respondToToolApproval() 返回用户的决定：

session.subscribe(event => {
  if (event.type === 'tool_approval_required') {
    session.respondToToolApproval({
      toolCallId: event.toolCallId,
      decision: 'approve',
    })
  }
})

always_allow_category 决定会为该门控所在 thread 授予工具类别。授权不是持久的进程级权限。

ask_user 和 submit_plan 等交互式工具使用可恢复的工具挂起。使用 session.respondToToolSuspension() 恢复它们：

session.subscribe(event => {
  if (event.type === 'tool_suspended' && event.toolName === 'ask_user') {
    void session.respondToToolSuspension({
      toolCallId: event.toolCallId,
      resumeData: 'Use SQLite.',
    })
  }
})

对于 submit_plan，使用 { action: 'approved' } 或 { action: 'rejected', feedback } 恢复。已批准的计划可以在运行继续之前切换到由 transitionsTo 配置的模式。

内置的 submit_plan 工具接收计划文件路径，而不是计划文本。Agent 将计划写入文件并提交其路径。它需要一个可写入的文件系统，例如 workspace 文件系统。要展示计划以供审批，你的应用应读取所提交 path 处的文件。

委派给 subagents

在 controller 上配置可用的 subagent 类型。随后，内置的 subagent 工具可以使用这些定义来委派聚焦任务：

const controller = new AgentController({
  tools: {
    searchDocs,
  },
  subagents: [
    {
      id: 'code-reviewer',
      name: 'Code reviewer',
      description: 'Review a change for correctness and regressions.',
      instructions: 'Inspect the change and report actionable findings.',
      allowedControllerTools: ['searchDocs'],
      allowedWorkspaceTools: ['view', 'find_files'],
      defaultModelId: 'openai/gpt-5-mini',
    },
  ],
})

常规 subagent 从其配置的指令和受限工具集开始。当子级应克隆父 thread 并使用父 agent 的指令和工具运行时，设置 forked: true。Forked subagents 会保留父提示前缀，忽略定义中的 instructions、tools、allowlists 和默认模型，并且要求 controller 上配置了 memory。

使用 session.subagents.model.set() 存储一个默认 subagent 模型，或为特定 agent 类型存储模型。使用 session.subagents.model.get() 读取选择：

await session.subagents.model.set({
  modelId: 'openai/gpt-5-mini',
})

await session.subagents.model.set({
  agentType: 'code-reviewer',
  modelId: 'anthropic/claude-sonnet-4-6',
})

const reviewerModel = session.subagents.model.get({
  agentType: 'code-reviewer',
})

这些选择会写入 thread 设置。Agent 类型的选择优先于 Session 的默认 subagent 模型。

连接聊天渠道

将渠道适配器传给 controller，并将其注册到 Mastra 实例上：

src/mastra/index.ts

import { Mastra } from '@mastra/core'
import { AgentController } from '@mastra/core/agent-controller'
import { createSlackAdapter } from '@chat-adapter/slack'

const controller = new AgentController({
  id: 'support-controller',
  agent,
  storage,
  workspace,
  modes,
  channels: {
    adapters: {
      slack: createSlackAdapter(),
    },
    resolveResourceId: ({ thread, message, defaultResourceId }) => {
      if (thread.isDM) return message.author.userId
      return defaultResourceId
    },
    onSessionStart: async ({ session, thread }) => {
      const plan = await billing.planFor(thread.resourceId)
      await session.model.switch(plan.modelId)
    },
  },
})

export const mastra = new Mastra({
  agentControllers: { controller },
  storage,
})

将每个平台 webhook 指向 controller 专用的路由：

/api/agent-controllers/<CONTROLLER_ID>/channels/<PLATFORM>/webhook

每个外部聊天 thread 映射到一个 controller Session 和一个 Mastra thread。默认情况下，新会话使用从适配器聊天 thread ID 派生的 resource ID，并带有 channel: 前缀。使用 resolveResourceId 将直接消息映射到现有应用用户，或选择另一个 memory 所有者。该回调只影响新 threads。现有 thread 会保留其已存储的 resource ID。

渠道会话由 controller 创建，而不是由你的代码创建，因此应在 onSessionStart 中配置它们。它每个会话运行一次，在会话绑定到其映射 thread 之后、处理第一条消息之前运行。渠道会话以 controller 默认值启动，因此这里是你设置其模型和 memory 设置的位置。同一 thread 中的后续消息会复用该会话，不会再次调用它。错误会被记录并吞掉，这样无法配置的会话仍能回答消息。

授权和路由渠道会话

onSessionStart 在会话存在后运行并吞掉错误，因此它无法拒绝请求。当宿主要决定会话是否可以存在时，使用 resolveSession。它会替换内置的会话创建逻辑，并在任何会话存在之前运行。抛出异常会在 controller 创建会话或调用模型之前拒绝请求。Mastra 会记录拒绝并让聊天 thread 保持沉默，因此你的授权消息永远不会到达渠道。

channels: {
  adapters: { slack: createSlackAdapter() },
  resolveSession: async ({ controller, thread, requestContext }) => {
    const install = await installs.authorize(requestContext.get('teamId'))

    return controller.createSession({
      resourceId: thread.resourceId,
      threadId: thread.id,
      scope: install.id,
      ownerId: controller.id,
      requestContext,
    })
  },
}

在 thread.resourceId 下创建会话。由于一个会话只能绑定它拥有的 threads，请使用 resolveResourceId 为映射 thread 分配不同的所有者。传入 threadId: thread.id，以便会话在映射 thread 上打开。否则，带 tags 或 projectPath 的会话不会将映射 thread 视为自己的，从而在切换到映射 thread 之前创建一个额外的空 thread。Sessions 按每个 resourceId 和 scope 组合执行 get-or-create。当一个 thread 需要为每个安装或主体提供单独会话时，传入 scope。

不是拒绝的失败（存储中断、解析器依赖中的 bug）仍会向 thread 发布错误，因此损坏的机器人不会看起来像是沉默的机器人。如果你需要在代码中区分它们，拒绝是一个 ChannelSessionRejectedError，其 cause 为原始错误。

当用户回答审批卡片时，resolveSession 也会使用该操作的 request context 运行。共享安装会重新验证批准人，而不是信任发送原始消息的人。

处理过期审批

审批门控存在于内存中，因此重启后回答的任何审批都已过期。Mastra 绝不会为过期操作运行工具。使用 onStaleToolApproval 来结算用户回答的尝试，而不是将其丢弃：

channels: {
  adapters: { slack: createSlackAdapter() },
  onStaleToolApproval: async ({ decision, toolCallId, runId, memory }) => {
    await runs.markInterrupted({ runId, toolCallId, decision, threadId: memory.thread })
  },
}

runId 是渲染审批卡片时所对应的运行，也就是用户回答的尝试，以及重启后你要结算的尝试。会话自身的运行会作为 currentRunId 单独传入，到那时它通常是 null 或另一个运行。

Controller 渠道会话和自动审批状态保存在内存中，因此请使用长期运行的服务器。待处理审批和实时 Session 状态不会在进程重启后保留。无法渲染审批控件的适配器会自动在没有审批提示的情况下运行工具，以免运行保持挂起。

关于适配器设置和平台特定的 webhook 配置，请参阅 Channels。

连接 UI

如何连接取决于 UI 运行的位置。终端 UI 或拥有 controller 的服务器持有 Session 对象并直接订阅它。浏览器 UI 在不同进程中运行，因此它通过 controller 的 HTTP 路由并使用 @mastra/client-js 访问同一会话。

服务器端会话

订阅 Session 事件以获取增量更新。当 UI 需要完整渲染快照时，使用 session.displayState.get() 读取简化后的展示状态：

const unsubscribe = session.subscribe(event => {
  if (event.type === 'display_state_changed') {
    render(event.displayState)
  }
})

render(session.displayState.get())

// Call when the UI disconnects.
unsubscribe()

订阅按 Session 隔离。同一 controller 上另一个 Session 的事件不会投递到此监听器。关于完整的 TUI 示例，请阅读构建编程 agent指南。

客户端会话

client.getAgentController(key).session(resourceId, scope?) 返回绑定到一个 resource 的会话客户端。传入 controller 在 new Mastra({ agentControllers }) 中注册时使用的 key，它可能与 controller 的 id 不同。Sessions 在服务器上执行 get-or-create，因此 create() 会恢复现有对话，而不是分叉它。当一个 resource 需要独立会话时（例如每个 git worktree 一个），传入 scope：

src/components/agent-session.ts

import { MastraClient } from '@mastra/client-js'

const client = new MastraClient({ baseUrl: 'http://localhost:4111' })
const session = client.getAgentController('coding-controller').session('user-123')

await session.create()

const subscription = await session.subscribe({
  onEvent: event => handleEvent(event),
  onError: error => showDisconnected(error),
  onReconnect: () => {
    void session.state().then(resync).catch(showDisconnected)
  },
  reconnect: true,
})

// Call when the UI disconnects.
subscription.unsubscribe()

与进程内监听器不同，subscribe() 接收选项对象并且是异步的。其 promise 在流建立后 resolve，在无法连接时 reject。被拒绝的调用不会在后台留下任何运行中的内容。reconnect: true 会在流建立后断开时使用指数退避重新建立流。如果连接在未关闭的情况下变得沉默（例如设备休眠后），当没有数据（包括服务器心跳）到达 idleTimeoutMs（默认 60000）时，会被视为断开。

服务器不会重放流中断期间错过的事件，因此在 onReconnect 中读取 session.state() 以获取当前模式、模型和 thread。客户端接口没有 session.displayState.get()。

使用 session.sendMessage(content) 发送工作，或使用 session.sendMessage({ content, files }) 附加 base64 编码的文件。回复会作为订阅上的 message_* 事件到达，而不是作为调用的返回值。使用 session.approveTool(toolCallId, approved) 回答 tool_approval_required 事件，使用 session.respondToToolSuspension(toolCallId, resumeData) 回答 tool_suspended 事件。

onEvent 会接收会话发出的每个事件。关于事件类型以及如何收窄它们，请参阅 Agent Controller API参考。

相关

• Agents
• Workspace
• Observational memory
• Channels
• Mastra Code
• Mastra Factory