客户端¶

SearchClient 是 AIMage SDK 的核心入口，负责与 AI-Mage Search API 的所有通信。当前版本只使用 ORPC 后端，所有高层方法最终都会调用 /rpc/<procedure>。

创建客户端¶

使用 search.client() 工厂函数创建客户端实例：

from aimage import search

client = search.client(
    token="your_token_here",
    service=SearchService.PROD,  # 可选，默认为生产环境
)

参数说明¶

参数	类型	默认值	说明
`token`	`str \| None`	`None`	Bearer Token。未传入时读取 `AIMAGE_SEARCH_TOKEN`
`api_key`	`str \| None`	`None`	API Key，优先级高于 `token`。未传入时读取 `AIMAGE_API_KEY`
`base_url`	`str \| None`	`None`	自定义 API origin。传入后会覆盖 `service`
`service`	`SearchService \\| str`	`SearchService.PROD`	服务环境，可传入枚举或自定义 URL
`**client_options`	`Any`	—	传递给 `SearchClient` 的额外选项（如 `timeout`、`headers` 等，最终传递给底层 `httpx.Client`）

服务环境¶

from aimage.search.settings import SearchService

# 使用预定义环境
client = search.client(token="...", service=SearchService.PROD)  # 生产环境
client = search.client(token="...", service=SearchService.DEV)  # 开发环境，仅开发者或受邀组织
client = search.client(token="...", service=SearchService.LOCAL)  # 本地环境

# 使用自定义 ORPC origin URL
client = search.client(token="...", service="https://custom-api.example.com")

# 也可以直接创建 SearchClient
from aimage.search.client import SearchClient

client = SearchClient(service=SearchService.DEV)

ORPC 与 Namespace¶

可以直接调用原始 ORPC，也可以使用对齐后端 router 的 namespace：

# 原始 ORPC
client.rpc("data/characters/list", {"project_id": "project_id"})

# Namespace
client.data.characters.list(project_id="project_id")
client.data.translationTables.entries.create(
    table_id="table_id",
    raw_values={"ja": "司", "en": "Tsukasa"},
)
client.kensaku.clips.updateSubtitle("clip_id", "new subtitle")
client.agent.read.projectContext(project_id="project_id")

主要 namespace：

Namespace	说明
`client.auth`	登录、用户信息、MFA / TOTP
`client.data`	角色、资源、参考图像、翻译表、漫画、小说、动态标签
`client.kensaku`	项目、视频、Clip、chatbot 对话
`client.agent`	Agent 读写接口

写入、更新、删除和资料管理接口见 CRUD 与更新接口。

邮件登录与 TOTP¶

进入 context 时会发送邮件验证码；拿到用户输入后调用 code()。

from aimage.search.client import SearchClient

client = SearchClient(service=SearchService.PROD)

with client.login_with_email(email="[email protected]") as login:
    code = input("email code> ")
    result = login.code(code)

如果后端返回需要 MFA，可以继续输入 TOTP：

with client.login_with_email(email="[email protected]") as login:
    login.code(input("email code> "))
    result = login.totp(input("totp> "))

也可以使用一次性调用：

client.login_with_email(
    email="[email protected]",
    email_code="123456",
    totp_code="654321",
)

Auth namespace¶

用途	Python 方法	Namespace
当前用户	`me()`	`client.auth.me()`
登出	`logout()`	`client.auth.logout()`
请求邮件验证码	`request_email_login_code(email)`	`client.auth.email.requestCode(email)`
验证邮件验证码	`verify_email_login_code(email, code)`	`client.auth.email.verifyCode(email, code)`
MFA 状态	`mfa_status()`	`client.auth.mfa.status()`
开始 TOTP 绑定	`mfa_enroll_start()`	`client.auth.mfa.enrollStart()`
确认 TOTP 绑定	`mfa_enroll_confirm(code)`	`client.auth.mfa.enrollConfirm(code)`
禁用 MFA	`mfa_disable(code)`	`client.auth.mfa.disable(code)`
重置恢复码	`mfa_regenerate_recovery_codes(code)`	`client.auth.mfa.regenerateRecoveryCodes(code)`
验证 MFA challenge	`verify_totp_challenge(challenge_token, code)`	`client.auth.mfa.verifyChallenge(challenge_token, code)`

SearchClient API¶

`health() -> bool`¶

检查 API 服务是否正常运行。

if client.health():
    print("服务正常")

如果服务异常，会抛出 RuntimeError 异常。

`projects() -> Generator[Project, None, None]`¶

获取当前用户可以访问的所有项目。返回一个生成器，支持惰性加载分页数据。

# 遍历所有项目
for project in client.projects():
    print(f"{project.name}: {project.description}")

# 转为列表
all_projects = list(client.projects())

`close() -> None`¶

关闭底层 HTTP 客户端连接。

client.close()

`client` 属性¶

访问底层的 httpx.Client 实例，用于高级定制。

http_client = client.client  # httpx.Client

上下文管理器¶

SearchClient 支持 with 语句，退出时自动关闭连接：

with search.client(token="...") as client:
    for project in client.projects():
        print(project.name)
# 连接已自动关闭

HTTP 配置¶

超时设置¶

默认超时配置为总超时 30 秒、连接超时 10 秒（httpx.Timeout(30.0, connect=10.0)）。

可以在创建客户端时直接传入自定义超时：

import httpx

client = search.client(
    token="...",
    timeout=httpx.Timeout(60.0, connect=20.0),
)

自定义请求头¶

client = search.client(
    token="...",
    headers={"X-Custom-Header": "value"},
)

认证机制¶

SDK 使用 Bearer Token 认证。创建客户端时传入的 token 会被设置为 Authorization: Bearer <token> 请求头。

Token 管理

Token 在调用 search.client() 时会被保存到全局设置中（settings.TOKEN），后续的内部 API 调用都会自动使用该 Token。

错误处理¶

SDK 内部使用 httpx 进行 HTTP 请求，所有响应都会调用 raise_for_status() 检查状态码。常见的异常包括：

异常	说明
`httpx.HTTPStatusError`	HTTP 请求返回非 2xx 状态码
`httpx.ConnectError`	无法连接到 API 服务
`httpx.TimeoutException`	请求超时
`RuntimeError`	健康检查失败

import httpx

try:
    client = search.client(token="invalid_token")
    projects = list(client.projects())
except httpx.HTTPStatusError as e:
    print(f"API 错误: {e.response.status_code}")
except httpx.ConnectError:
    print("无法连接到 API 服务")