name: chatkit-botbuilder description: 创建生产级ChatKit聊天机器人的指南，该机器人集成了OpenAI Agents SDK、MCP工具和自定义后端。在构建具有专业能力、实时任务执行和用户隔离功能的AI驱动聊天机器人时使用。 license: 完整条款见LICENSE.txt
ChatKit Botbuilder
概述
使用OpenAI ChatKit框架创建生产级聊天机器人。此技能支持构建能够：
- 集成AI智能体：使用OpenAI Agents SDK进行智能对话处理
- 执行工具：连接MCP（模型上下文协议）工具以执行现实世界任务
- 支持自定义后端：构建具有完整协议支持的FastAPI后端
- 确保用户隔离：通过JWT身份验证实现多用户系统
- 实时同步：当聊天机器人执行操作时启用实时UI更新
- 灵活部署：部署到Web、移动或桌面应用程序
此技能提供了ChatKit集成的完整架构模式，从前端配置到后端服务器实现。
何时使用此技能
在以下情况下使用此技能：
- 构建任务管理聊天机器人 - 为任务创建、更新、完成创建对话界面
- 将AI集成到现有应用程序 - 将ChatKit添加到仪表板、Web应用程序或平台
- 创建专业AI助手 - 构建具有自定义工具集成的领域特定聊天机器人
- 实现多用户聊天机器人 - 创建每个用户拥有隔离对话和数据的系统
- 添加实时功能 - 构建能够触发实际应用程序更改的聊天机器人
- 部署AI对话 - 创建与您的数据库和API交互的聊天机器人
架构概述
高层流程
用户消息
↓
ChatKit前端 (React/Next.js)
↓ [Authorization Header中的JWT Token]
↓
FastAPI后端 (ChatKit服务器)
↓ [从JWT提取user_id]
↓
OpenAI智能体 (Agents SDK)
↓ [需要工具执行]
↓
MCP工具 (自定义工具函数)
↓ [创建/更新/列出数据]
↓
数据库 (用户隔离数据)
↓
响应 → ChatKit → 前端 → 用户
关键组件
-
前端 (Next.js + ChatKit SDK)
- 具有对话历史的ChatKit UI组件
- localStorage中的JWT令牌管理
- 带有Bearer令牌身份验证的自定义fetch包装器
- 与后端更改同步的实时自动刷新
-
后端 (FastAPI + ChatKit服务器)
- 处理请求的ChatKit协议端点
- 扩展ChatKitServer的MyChatKitServer类
- 通过JWT中间件实现用户隔离
- 用于自动注入user_id的工具包装函数
-
智能体 (OpenAI Agents SDK)
- 带有指令的任务管理智能体
- 工具注册和执行
- 会话管理
-
工具 (MCP + 自定义函数)
- 自动注入user_id的包装函数
- 具有用户隔离的数据库操作
- 一致的错误处理
-
数据库
- SQLModel ORM模型
- 按用户任务过滤
- 对话持久化
快速入门工作流
阶段1：后端设置 (FastAPI)
1. 创建ChatKit服务器类
from chatkit.server import ChatKitServer
from chatkit.store import Store
class MyChatKitServer(ChatKitServer):
def __init__(self):
store = CustomChatKitStore()
super().__init__(store=store)
async def respond(self, thread, input, context):
"""处理用户消息并流式传输AI响应"""
user_id = getattr(context, 'user_id', None)
# 使用包装工具创建智能体
# 使用官方模式流式传输响应
2. 创建MCP工具包装器
# 从上下文提取user_id并注入工具调用
def add_task_wrapper(title: str, description: str = None):
return mcp_add_task(user_id=user_id, title=title, description=description)
def list_tasks_wrapper(status: str = "all"):
return mcp_list_tasks(user_id=user_id, status=status)
3. 创建FastAPI端点
@router.post("/api/v1/chatkit")
async def chatkit_protocol_endpoint(request: Request):
user_id = request.state.user_id # 来自JWT中间件
context = create_context_object(user_id=user_id)
result = await chatkit_server.process(body, context)
return StreamingResponse(result, media_type="text/event-stream")
4. 配置JWT中间件
class JWTAuthMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request, call_next):
# 从Authorization头提取JWT令牌
# 解码并设置request.state.user_id
# 所有端点都可以访问经过身份验证的user_id
阶段2：前端设置 (Next.js + React)
1. 配置ChatKit SDK
const chatKitConfig: UseChatKitOptions = {
api: {
url: `${API_BASE_URL}/api/v1/chatkit`,
domainKey: 'your-domain-key',
fetch: authenticatedFetch, // 带有JWT的自定义fetch
},
theme: 'light',
header: { enabled: true, title: { text: 'AI聊天' } },
history: { enabled: true },
}
2. 创建经过身份验证的Fetch包装器
async function authenticatedFetch(input, options) {
const token = localStorage.getItem('access_token')
const headers = {
...options?.headers,
'Authorization': `Bearer ${token}`,
}
return fetch(input, { ...options, headers })
}
3. 集成ChatKit小部件
import { ChatKitWidget } from '@openai/chatkit-react'
export default function Dashboard() {
return (
<div className="flex gap-4">
{/* 您的应用内容 */}
{showChat && (
<ChatKitWidget {...chatKitConfig} />
)}
</div>
)
}
4. 为实时同步添加自动刷新
useEffect(() => {
if (!showChatKit) return
// 聊天打开时立即刷新
fetchTasks()
// 每1秒刷新一次以实现实时更新
const interval = setInterval(() => {
fetchTasks()
}, 1000)
return () => clearInterval(interval)
}, [showChatKit])
阶段3：工具实现 (MCP)
1. 创建具有用户隔离的MCP工具
def add_task(user_id: str, title: str, description: Optional[str] = None):
"""创建任务 - 从包装器接收user_id"""
task = Task(
id=str(uuid.uuid4()),
user_id=user_id, # 关键：确保用户隔离
title=title,
description=description,
completed=False,
created_at=datetime.utcnow(),
)
with Session(engine) as session:
session.add(task)
session.commit()
2. 注册MCP工具
mcp_server = MCPServer()
mcp_server.register_tool("add_task", add_task)
mcp_server.register_tool("list_tasks", list_tasks)
mcp_server.register_tool("delete_task", delete_task)
# ... 更多工具
核心模式与最佳实践
1. 用户隔离策略
三级保证：
- 中间件级别 - JWT验证确保仅经过身份验证的用户
- 工具级别 - 包装器函数自动注入user_id
- 数据库级别 - 所有查询按user_id过滤
# 中间件从令牌提取user_id
request.state.user_id = payload.get("user_id")
# 工具包装器捕获并注入它
def add_task_wrapper(title):
return mcp_add_task(user_id=user_id, ...)
# 数据库强制执行它
WHERE user_id = ? AND task_id = ?
2. 带有用户上下文的消息流
用户发送："创建一个名为'买牛奶'的任务"
↓
ChatKit协议：POST /api/v1/chatkit
头：Authorization: Bearer <JWT>
体：{ "type": "message", "text": "Create..." }
↓
JWT中间件：
从令牌提取user_id → request.state.user_id
↓
ChatKit服务器 (MyChatKitServer.respond)：
从上下文获取user_id
创建捕获user_id的包装器函数
将包装器传递给智能体
↓
OpenAI智能体：
接收消息："创建一个任务..."
选择工具：add_task_wrapper
调用：add_task_wrapper(title="买牛奶")
↓
包装器函数：
调用：mcp_add_task(user_id="user-123", title="买牛奶")
↓
MCP工具：
使用正确的user_id创建任务
返回：{"task_id": "...", "title": "买牛奶"}
↓
智能体响应：
"我已创建'买牛奶'任务 ✓"
↓
ChatKit前端：
显示响应
自动刷新任务列表 → 看到新任务
3. 流式响应模式
# 使用Runner.run_streamed的官方ChatKit模式
result = Runner.run_streamed(
task_agent.agent,
agent_input,
context=agent_context,
)
# 使用官方stream_agent_response流式传输事件
async for event in stream_agent_response(agent_context, result):
yield event
4. 线程和项目管理
# 将用户消息添加到线程
await self.store.add_thread_item(thread.id, input, context)
# 加载对话历史
items_page = await self.store.load_thread_items(
thread.id,
after=None,
limit=30,
order="desc",
context=context,
)
# 转换为智能体输入
agent_input = await simple_to_agent_input(items)
集成模式
模式1：任务管理聊天机器人 (基础)
功能：
- 用户通过与ChatKit对话创建任务
- ChatKit在侧边栏显示任务列表
- 自动刷新保持任务列表同步
参考文件：
模式2：多应用ChatKit部署
功能：
- 将ChatKit部署到多个应用程序
- 共享相同的后端和数据库
- 每个应用程序具有隔离的用户上下文
关键设置：
- 使用环境变量设置API_BASE_URL
- 为每个应用程序配置域密钥
- 实现每个应用程序的身份验证
模式3：实时协作
功能：
- 多个用户与同一聊天机器人实例聊天
- 自动刷新保持每个人的数据同步
- 用户隔离防止跨用户数据泄露
实现：
- 用于真正实时的WebSocket连接 (可选高级功能)
- 使用自动刷新的轮询简化实现
- 用于数据一致性的数据库事务
常见问题与解决方案
问题1：在ChatKit中创建的任务未出现在仪表板中
根本原因： user_id未传递给MCP工具
解决方案： 使用捕获并注入user_id的包装器函数
def add_task_wrapper(title):
return mcp_add_task(user_id=user_id, title=title)
问题2：一个用户看到另一个用户的任务
根本原因： 数据库查询中缺少user_id过滤器
解决方案： 始终在工具级别按user_id过滤
stmt = select(Task).where(
Task.user_id == user_id,
Task.completed == False
)
问题3：找不到ChatKit API端点
根本原因： 路由器未包含在FastAPI应用中
解决方案： 在main.py中包含路由器
from routes import chatkit
app.include_router(chatkit.router)
问题4：聊天小部件未显示消息
根本原因： 自定义fetch未添加JWT令牌
解决方案： 确保authenticatedFetch添加Bearer令牌
const token = localStorage.getItem('access_token')
headers['Authorization'] = `Bearer ${token}`
高级主题
实时更新 (WebSocket)
用于真正实时 (非轮询)：
- 在HTTP端点旁边实现WebSocket端点
- 向所有连接的客户端广播更新
- 使用用户上下文维护连接状态
自定义工具模式
为ChatKit小部件构建工具响应：
return {
"tasks": task_list,
"total": len(task_list),
"pending": pending_count,
"message": "您有5个任务",
"widget": {
"type": "card",
"items": formatted_items,
}
}
会话持久化
在数据库中存储对话历史：
- 将对话链接到用户
- 为上下文检索聊天历史
- 允许恢复对话
资源
此技能包含构建ChatKit聊天机器人的全面资源：
references/
后端架构： 完整的FastAPI ChatKit服务器实现细节和模式
前端集成： Next.js ChatKit小部件配置和身份验证
MCP工具指南： 创建具有自动user_id注入的包装工具函数
用户隔离： 三级用户隔离策略和验证清单
scripts/
chatkit_server_template.py - 具有所有必需方法的FastAPI ChatKit服务器样板
mcp_wrapper_generator.py - 自动生成MCP工具包装器的脚本
frontend_config_generator.ts - 用于ChatKit前端设置的TypeScript配置生成器
assets/
chatkit-nextjs-template/ - 集成了ChatKit的完整Next.js项目
fastapi-backend-template/ - 具有ChatKit服务器实现的完整FastAPI后端
验证清单
构建ChatKit聊天机器人时，验证：
- [ ] JWT中间件从令牌提取user_id
- [ ] ChatKit端点在上下文中接收user_id
- [ ] 工具包装器捕获并注入user_id
- [ ] 数据库查询按user_id过滤
- [ ] 前端authenticatedFetch包含Bearer令牌
- [ ] ChatKit配置指向正确的后端端点
- [ ] 自动刷新定期获取更新的数据
- [ ] 一个用户无法看到另一个用户的数据
- [ ] 聊天机器人可以成功调用MCP工具
- [ ] 工具响应出现在ChatKit对话中
- [ ] 聊天机器人和仪表板之间的实时同步有效
后续步骤
- 对于新项目： 从
assets/fastapi-backend-template/和assets/chatkit-nextjs-template/复制模板 - 对于现有应用： 遵循“集成模式”部分并参考架构指南
- 对于高级功能： 阅读“高级主题”部分并根据需要扩展
- 对于故障排除： 检查“常见问题与解决方案”并验证清单