跳转到内容
Imbrace SDK

集成

两个 SDK 的框架级接线模式。TypeScript 部分涵盖 React、Next.js 和 Node.js；Python 部分涵盖 FastAPI、asyncio、Django 和 Celery。OTP 登录流程为两者均有记录。

凭证策略（api key vs access token、env vars），参阅身份验证和安装指南。

React (TypeScript)

单例客户端

在组件树外创建一次客户端并复用。localStorage 中的 token 来自 OTP 登录流程。

lib/imbrace.ts
import { ImbraceClient } from "@imbrace/sdk";


export const client = new ImbraceClient({
  accessToken:
    typeof window !== "undefined"
      ? (localStorage.getItem("imbrace_token") ?? undefined)
      : undefined,
});

数据获取 Hook

hooks/useProducts.ts
import { useState, useEffect } from "react";
import { client } from "@/lib/imbrace";
import type { Product } from "@imbrace/sdk";


export function useProducts(category?: string) {
  const [products, setProducts] = useState<Product[]>([]);
  const [loading, setLoading] = useState(true);
  const [error, setError] = useState<Error | null>(null);


  useEffect(() => {
    setLoading(true);
    client.marketplace
      .listProducts({ category })
      .then((res) => setProducts(res.data))
      .catch(setError)
      .finally(() => setLoading(false));
  }, [category]);


  return { products, loading, error };
}
components/ProductList.tsx
import { useProducts } from "@/hooks/useProducts";


export function ProductList() {
  const { products, loading, error } = useProducts("electronics");
  if (loading) return <p>加载中...</p>;
  if (error) return <p>错误：{error.message}</p>;
  return (
    <ul>
      {products.map((p) => (
        <li key={p.id}>{p.name} — {p.price} {p.currency}</li>
      ))}
    </ul>
  );
}

Next.js (TypeScript)

API 路由（App Router）

app/api/products/route.ts
import { NextResponse } from "next/server";
import { ImbraceClient } from "@imbrace/sdk";


const client = new ImbraceClient({
  apiKey: process.env.IMBRACE_API_KEY,
});


export async function GET(request: Request) {
  const { searchParams } = new URL(request.url);
  const category = searchParams.get("category") ?? undefined;
  const { data } = await client.marketplace.listProducts({ category });
  return NextResponse.json(data);
}


export async function POST(request: Request) {
  const body = await request.json();
  const product = await client.marketplace.createProduct(body);
  return NextResponse.json(product, { status: 201 });
}

process.env.IMBRACE_API_KEY 应按你的部署平台的方式设置。参阅安装指南 → 配置凭证。

服务端组件（App Router）

app/products/page.tsx
import { ImbraceClient } from "@imbrace/sdk";


const client = new ImbraceClient({
  apiKey: process.env.IMBRACE_API_KEY,
});


export default async function ProductsPage() {
  const { data: products } = await client.marketplace.listProducts({ limit: 20 });
  return (
    <main>
      <h1>商品</h1>
      <ul>{products.map((p) => <li key={p.id}>{p.name}</li>)}</ul>
    </main>
  );
}

Node.js CLI 脚本 (TypeScript)

适用于一次性脚本（数据导出、回填、临时查询）：

scripts/export-contacts.ts
import { ImbraceClient } from "@imbrace/sdk";
import { writeFileSync } from "fs";


const client = new ImbraceClient();


async function exportContacts() {
  const { data: contacts } = await client.contacts.list({ limit: 1000 });
  writeFileSync("contacts.json", JSON.stringify(contacts, null, 2));
  console.log(`已导出 ${contacts.length} 个联系人`);
}


exportContacts().catch(console.error);
Terminal window
npx ts-node scripts/export-contacts.ts

FastAPI (Python)

Per-request 依赖注入

最简单的模式 — 每个请求一个 async client，由依赖项管理生命周期：

from fastapi import FastAPI, Depends
from imbrace import AsyncImbraceClient


app = FastAPI()


async def get_imbrace() -> AsyncImbraceClient:
    async with AsyncImbraceClient() as client:
        yield client


@app.get("/products")
async def list_products(client: AsyncImbraceClient = Depends(get_imbrace)):
    result = await client.marketplace.list_products(limit=20)
    return result["data"]


@app.get("/products/{product_id}")
async def get_product(product_id: str, client: AsyncImbraceClient = Depends(get_imbrace)):
    return await client.marketplace.get_product(product_id)


@app.post("/ai/chat")
async def chat(message: str, client: AsyncImbraceClient = Depends(get_imbrace)):
    return await client.ai.complete(
        model="gpt-4o",
        messages=[{"role": "user", "content": message}],
    )

全局单例（提高连接复用）

对于更高吞吐量，在应用程序生命周期内共享一个 client：

from contextlib import asynccontextmanager
from fastapi import FastAPI
from imbrace import AsyncImbraceClient


imbrace: AsyncImbraceClient = None  # type: ignore


@asynccontextmanager
async def lifespan(app: FastAPI):
    global imbrace
    imbrace = AsyncImbraceClient()
    await imbrace.init()  # 启动时健康检查
    yield
    await imbrace.close()


app = FastAPI(lifespan=lifespan)


@app.get("/me")
async def get_me():
    return await imbrace.platform.get_me()

asyncio (Python)

并发请求

import asyncio
from imbrace import AsyncImbraceClient


async def fetch_dashboard_data():
    async with AsyncImbraceClient() as client:
        me, products, channels = await asyncio.gather(
            client.platform.get_me(),
            client.marketplace.list_products(limit=5),
            client.channel.list_channels(type="group"),
        )
        return {
            "user": me,
            "products": products["data"],
            "channels": channels["data"],
        }


data = asyncio.run(fetch_dashboard_data())

流式 AI

import asyncio
from imbrace import AsyncImbraceClient


async def stream_response():
    async with AsyncImbraceClient() as client:
        async for chunk in client.ai.stream(
            model="gpt-4o",
            messages=[{"role": "user", "content": "解释 Python 中的 async/await。"}],
        ):
            content = chunk["choices"][0]["delta"].get("content", "")
            print(content, end="", flush=True)


asyncio.run(stream_response())

Django (Python)

同步视图

views.py
from django.http import JsonResponse
from imbrace import ImbraceClient, ApiError


def product_list(request):
    with ImbraceClient() as client:
        try:
            result = client.marketplace.list_products(
                category=request.GET.get("category"),
                page=int(request.GET.get("page", 1)),
            )
            return JsonResponse(result)
        except ApiError as e:
            return JsonResponse({"error": str(e)}, status=e.status_code)

Django 设置集成

settings.py
IMBRACE_API_KEY = env("IMBRACE_API_KEY")
IMBRACE_ENV = env("IMBRACE_ENV", default="stable")


# utils/imbrace.py
from django.conf import settings
from imbrace import ImbraceClient


def get_client() -> ImbraceClient:
    return ImbraceClient(
        api_key=settings.IMBRACE_API_KEY,
        env=settings.IMBRACE_ENV,
    )

Celery (Python)

对于后台任务 workers，在每个 task 内部创建 client — 不要在 worker 之间共享：

tasks.py
from celery import Celery
from imbrace import ImbraceClient, NetworkError


app = Celery("tasks")


@app.task(bind=True, max_retries=3)
def sync_products(self):
    try:
        with ImbraceClient() as client:
            result = client.marketplace.list_products(limit=100)
            for product in result["data"]:
                save_to_db(product)
    except NetworkError as exc:
        raise self.retry(exc=exc, countdown=2 ** self.request.retries)

OTP 登录流程

OTP 流程在两个 SDK 中概念上相同：为邮箱请求 OTP，然后交换为 access token。参阅身份验证 → OTP 登录流程了解完整的凭证生命周期。

components/LoginForm.tsx
import { useState } from "react";
import { ImbraceClient, AuthError } from "@imbrace/sdk";


const client = new ImbraceClient();


export function LoginForm() {
  const [email, setEmail] = useState("");
  const [otp, setOtp] = useState("");
  const [step, setStep] = useState<"email" | "otp">("email");


  async function requestOtp() {
    await client.requestOtp(email);
    setStep("otp");
  }


  async function verifyOtp() {
    try {
      await client.loginWithOtp(email, otp);
      window.location.href = "/dashboard";
    } catch (e) {
      if (e instanceof AuthError) alert("OTP 不正确");
    }
  }


  return step === "email" ? (
    <div>
      <input value={email} onChange={(e) => setEmail(e.target.value)} placeholder="邮箱" />
      <button onClick={requestOtp}>发送 OTP</button>
    </div>
  ) : (
    <div>
      <input value={otp} onChange={(e) => setOtp(e.target.value)} placeholder="输入 OTP" />
      <button onClick={verifyOtp}>确认</button>
    </div>
  );
}