⚡

构建聊天界面Skill building-chat-interfaces

本技能专注于构建生产级的AI聊天界面，提供完整的全栈解决方案。涵盖前端React组件集成、后端Python服务器开发、身份验证机制（JWT/JWKS）、上下文注入、数据库持久化、SSE流式响应、Next.js集成以及MCP工具认证等核心功能。适用于需要将AI智能体嵌入到Web应用、实现个性化对话体验、管理用户会话历史的企业级应用场景。关键词：AI聊天界面、ChatKit集成、全栈开发、身份验证、上下文注入、流式响应、Next.js代理、MCP工具、生产级部署。

前端开发 0 次安装 5 次浏览 更新于 3/2/2026

名称：构建聊天界面 描述：| 使用自定义后端、身份验证和上下文注入构建AI聊天界面。 适用于将聊天UI与AI智能体集成、为聊天添加身份验证、注入用户/页面上下文、 或实现httpOnly cookie代理的场景。涵盖ChatKitServer、useChatKit和MCP身份验证模式。 不适用于构建没有持久化或自定义智能体集成的简单聊天机器人。

构建聊天界面

构建生产级的AI聊天界面，支持自定义后端集成。

快速开始

# 后端（Python）
uv add chatkit-sdk agents httpx

# 前端（React）
npm install @openai/chatkit-react

核心架构

前端（React）                    后端（Python）
┌─────────────────┐                ┌─────────────────┐
│  useChatKit()   │───HTTP/SSE───>│  ChatKitServer  │
│  - 自定义fetch   │                │  - respond()    │
│  - 身份验证头    │                │  - 存储         │
│  - 页面上下文    │                │  - 智能体       │
└─────────────────┘                └─────────────────┘

后端模式

1. 带自定义智能体的ChatKit服务器

from chatkit.server import ChatKitServer
from chatkit.agents import stream_agent_response
from agents import Agent, Runner

class CustomChatKitServer(ChatKitServer[RequestContext]):
    """使用自定义智能体扩展ChatKit服务器。"""

    async def respond(
        self,
        thread: ThreadMetadata,
        input_user_message: UserMessageItem | None,
        context: RequestContext,
    ) -> AsyncIterator[ThreadStreamEvent]:
        if not input_user_message:
            return

        # 加载对话历史
        previous_items = await self.store.load_thread_items(
            thread.id, after=None, limit=10, order="desc", context=context
        )

        # 为提示构建历史字符串
        history_str = "
".join([
            f"{item.role}: {item.content}"
            for item in reversed(previous_items.data)
        ])

        # 从元数据中提取上下文
        user_info = context.metadata.get('userInfo', {})
        page_context = context.metadata.get('pageContext', {})

        # 在指令中创建带上下文的智能体
        agent = Agent(
            name="助手",
            tools=[your_search_tool],
            instructions=f"{history_str}
用户: {user_info.get('name')}
{system_prompt}",
        )

        # 运行带流式输出的智能体
        result = Runner.run_streamed(agent, input_user_message.content)
        async for event in stream_agent_response(context, result):
            yield event

2. 数据库持久化

from sqlmodel.ext.asyncio.session import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine

DATABASE_URL = os.getenv("DATABASE_URL").replace("postgresql://", "postgresql+asyncpg://")
engine = create_async_engine(DATABASE_URL, pool_pre_ping=True)

# 启动时预热连接池
async def warmup_pool():
    async with engine.begin() as conn:
        await conn.execute(text("SELECT 1"))

3. JWT/JWKS身份验证

from jose import jwt
import httpx

async def get_current_user(authorization: str = Header()):
    token = authorization.replace("Bearer ", "")
    async with httpx.AsyncClient() as client:
        jwks = (await client.get(JWKS_URL)).json()
    payload = jwt.decode(token, jwks, algorithms=["RS256"])
    return payload

前端模式

1. 自定义Fetch拦截器

const { control, sendUserMessage } = useChatKit({
  api: {
    url: `${backendUrl}/chatkit`,
    domainKey: domainKey,

    // 自定义fetch以注入身份验证和上下文
    fetch: async (url: string, options: RequestInit) => {
      if (!isLoggedIn) {
        throw new Error('用户必须登录');
      }

      const pageContext = getPageContext();
      const userInfo = { id: userId, name: user.name };

      // 将元数据注入请求体
      let modifiedOptions = { ...options };
      if (modifiedOptions.body && typeof modifiedOptions.body === 'string') {
        const parsed = JSON.parse(modifiedOptions.body);
        if (parsed.params?.input) {
          parsed.params.input.metadata = {
            userId, userInfo, pageContext,
            ...parsed.params.input.metadata,
          };
          modifiedOptions.body = JSON.stringify(parsed);
        }
      }

      return fetch(url, {
        ...modifiedOptions,
        headers: {
          ...modifiedOptions.headers,
          'X-User-ID': userId,
          'Content-Type': 'application/json',
        },
      });
    },
  },
});

2. 页面上下文提取

const getPageContext = useCallback(() => {
  if (typeof window === 'undefined') return null;

  const metaDescription = document.querySelector('meta[name="description"]')
    ?.getAttribute('content') || '';

  const mainContent = document.querySelector('article') ||
                     document.querySelector('main') ||
                     document.body;

  const headings = Array.from(mainContent.querySelectorAll('h1, h2, h3'))
    .slice(0, 5)
    .map(h => h.textContent?.trim())
    .filter(Boolean)
    .join(', ');

  return {
    url: window.location.href,
    title: document.title,
    path: window.location.pathname,
    description: metaDescription,
    headings: headings,
  };
}, []);

3. 脚本加载检测

const [scriptStatus, setScriptStatus] = useState<'pending' | 'ready' | 'error'>(
  isBrowser && window.customElements?.get('openai-chatkit') ? 'ready' : 'pending'
);

useEffect(() => {
  if (!isBrowser || scriptStatus !== 'pending') return;

  if (window.customElements?.get('openai-chatkit')) {
    setScriptStatus('ready');
    return;
  }

  customElements.whenDefined('openai-chatkit').then(() => {
    setScriptStatus('ready');
  });
}, []);

// 仅在就绪时渲染
{isOpen && scriptStatus === 'ready' && <ChatKit control={control} />}

Next.js集成

httpOnly Cookie代理

当身份验证令牌位于httpOnly cookie中时（JavaScript无法读取）：

// app/api/chatkit/route.ts
import { NextRequest, NextResponse } from "next/server";
import { cookies } from "next/headers";

export async function POST(request: NextRequest) {
  const cookieStore = await cookies();
  const idToken = cookieStore.get("auth_token")?.value;

  if (!idToken) {
    return NextResponse.json({ error: "未认证" }, { status: 401 });
  }

  const response = await fetch(`${API_BASE}/chatkit`, {
    method: "POST",
    headers: {
      Authorization: `Bearer ${idToken}`,
      "Content-Type": "application/json",
    },
    body: await request.text(),
  });

  // 处理SSE流式传输
  if (response.headers.get("content-type")?.includes("text/event-stream")) {
    return new Response(response.body, {
      status: response.status,
      headers: {
        "Content-Type": "text/event-stream",
        "Cache-Control": "no-cache",
      },
    });
  }

  return NextResponse.json(await response.json(), { status: response.status });
}

脚本加载策略

// app/layout.tsx
import Script from "next/script";

export default function RootLayout({ children }: { children: React.ReactNode }) {
  return (
    <html lang="en">
      <head>
        {/* 对于Web组件，必须使用beforeInteractive */}
        <Script
          src="https://cdn.platform.openai.com/deployments/chatkit/chatkit.js"
          strategy="beforeInteractive"
        />
      </head>
      <body>{children}</body>
    </html>
  );
}

MCP工具身份验证

MCP协议不转发身份验证头。通过系统提示传递凭据：

SYSTEM_PROMPT = """你是助手。

## 身份验证上下文
- 用户ID: {user_id}
- 访问令牌: {access_token}

关键：调用任何MCP工具时，请包含：
- user_id: "{user_id}"
- access_token: "{access_token}"
"""

# 使用凭据格式化
instructions = SYSTEM_PROMPT.format(
    user_id=context.user_id,
    access_token=context.metadata.get("access_token", ""),
)

常见问题

问题 症状 解决方案
历史记录不在提示中 智能体不记得对话 在系统提示中包含历史记录字符串
上下文未传输 智能体缺少用户/页面信息 添加到请求元数据中，在后端提取
脚本未加载 组件渲染失败 检测脚本加载状态，等待渲染
缺少身份验证头 后端拒绝请求 使用自定义fetch拦截器
httpOnly cookie 无法从JS读取令牌 创建服务器端API路由代理
首次请求缓慢 7+秒延迟 预热数据库连接池

验证

运行：python3 scripts/verify.py

预期结果：✓ building-chat-interfaces 技能就绪

如果验证失败

  1. 检查：references/文件夹中有chatkit-integration-patterns.md
  2. 如果仍然失败，请停止并报告

相关技能（分层系统）

  • streaming-llm-responses - 第2层：响应生命周期、进度更新、客户端效果
  • building-chat-widgets - 第3层：交互式小部件、实体标记、编辑器工具
  • fetching-library-docs - ChatKit文档：--library-id /openai/chatkit --topic useChatKit

参考资料