Python 方法与类型

本文按当前 SDK 源码的公开导出和方法签名说明 jusp-platform-sdk。分发包名是 jusp-platform-sdk，导入名是 jusp_sdk；安装版本以应用开发者工作台私有索引显示为准。不包含 Agent Runtime SDK。

媒体生成的完整上传、优化、创建、恢复、查询和下载链路请看媒体生成完整调用链。

先看三个运行模式

RuntimeHttpClient 没有 debug token 时进入 deployed 模式；此时必须有 session ID 和 ServiceAccount 凭据。三种模式使用相同的命名调用方式，但媒体快捷方法和正式 Job 接口 不是同一条 HTTP 路径。

通用传输、错误与幂等

同步客户端使用 httpx.Client，异步客户端使用 httpx.AsyncClient。默认连接池最多 100 个连接、20 个保活连接、保活 30 秒；默认总体超时 30 秒、连接超时 5 秒。 注入的 transport/client 由调用方关闭，SDK 自建的客户端由 close() 关闭。

SDK 不自动重放没有幂等键的非幂等请求。异步创建、提示词优化和应用间授权应使用稳定的 业务幂等键；服务端返回的 429 退避信息由 RuntimeAPIError.retry_after_seconds 提供。

RuntimeClient：会话与组织目录

RuntimeClient(
    *, app_key: str,
    base_url: str | None = None,
    sa_token: str | None = None,
    sa_token_file: str | None = None,
    access_token: str | None = None,
    app_id: str | None = None,
    app_version_id: str | None = None,
    developer_base_url: str | None = None,
    transport: httpx.Client | None = None,
    max_retries: int = 0,
    timeout: float | httpx.Timeout = DEFAULT_TIMEOUT,
    limits: httpx.Limits = DEFAULT_LIMITS,
    token_refresh_interval_seconds: float = 1.0,
    resolve_cache_ttl_seconds: float = 1.0,
    resolve_cache_max_entries: int = 2048,
)

同步公开方法：

示例：

from jusp_sdk.runtime import RuntimeClient

sessions = RuntimeClient(app_key="your-app")
try:
    session = sessions.exchange(request_json["launchCode"])
    session = sessions.renew_if_needed(session)
    identity = session.identity()
    result = {"accountId": identity.account_id, "roles": list(identity.roles.roles)}
finally:
    sessions.close()

RuntimeSession 是不可变快照，字段为 session_id、app_key、app_version_id、 subject、app_roles、expires_at。RuntimeSubject 必填 tenant_id/account_id/workspace_id/user_id，可选 organization_id/membership_id。 应用不得接受浏览器提交的身份字段覆盖解析结果。

AsyncRuntimeClient

AsyncRuntimeClient 构造参数与 RuntimeClient 相同，但传入的是 httpx.AsyncClient（参数名 transport）。exchange、resolve、renew、 renew_if_needed、组织目录方法和 close 需要 await；identity(session) 是同步纯本地 方法。相同 session 的解析和相同旧 session 的续期分别合并为一个异步任务；取消一个等待者 不会取消其他等待者。

from jusp_sdk.runtime import AsyncRuntimeClient

sessions = AsyncRuntimeClient(app_key="your-app")
try:
    session = await sessions.exchange(launch_code)
finally:
    await sessions.close()

RuntimeHttpClient：统一 Runtime 调用

RuntimeHttpClient(
    *, base_url: str | None = None,
    access_token: str | None = None,
    sa_token: str | None = None,
    sa_token_file: str | None = None,
    session: RuntimeSession | None = None,
    session_id: str | None = None,
    client: httpx.Client | None = None,
    timeout: float | httpx.Timeout = DEFAULT_TIMEOUT,
    limits: httpx.Limits = DEFAULT_LIMITS,
    token_refresh_interval_seconds: float = 1.0,
    dev_scenario: str | None = None,
)

AsyncRuntimeHttpClient 参数相同，但 client 类型为 httpx.AsyncClient，网络方法和 close() 需要 await。两者公开方法如下：

动态快捷方法由 __getattr__ 提供，准确调用形状是 payload, **kwargs：

reply = client.chat_completions({"model": "model-alias", "messages": [{"role": "user", "content": "hello"}]})
vectors = client.embeddings({"model": "model-alias", "input": ["hello"]})

快捷名称与 operation 的对应关系是：chat_completions → chat、embeddings → embeddings、 rerank → rerank、ocr → ocr、audio_transcriptions → audio_transcriptions、 image_generation → image_generation、video_generation → video_generation、 audio_generation → audio_generation、document → document。异步客户端的动态方法返回 awaitable。ASR 是例外：公网入口要求 multipart/form-data 的 file，而当前 SDK 的 audio_transcriptions(payload) 是 JSON 快捷方法，不能把 bytes 放进 audio 字段后照抄， 否则会因 JSON 不可序列化或被接口拒绝。公网 Python 请求应直接使用 multipart：

import httpx

with open("speech.mp3", "rb") as audio:
    response = httpx.post(
        f"{API_BASE}/audio/transcriptions",
        headers={"Authorization": f"Bearer {API_TOKEN}"},
        data={"model": MODEL_ALIAS},
        files={"file": ("speech.mp3", audio, "audio/mpeg")},
        timeout=30,
    )
    response.raise_for_status()
    transcription = response.json()

已部署应用才使用 SDK JSON 入口，并传 inputs: [{"readUrl": ...}] 的正式 Runtime 合同。SDK 不负责音频编解码、分段、说话人分离或语音内容后处理。

RuntimeDownload、Job 与素材字段

RuntimeDownload 的真实字段只有：

它没有 content_length 或 etag 字段。head_asset() 返回的 dict 才包含 etag。 RuntimeJob 字段为 job_id/client_task_id/status/stage/recovery_ref/results；每个 RuntimeResult 有 result_id/download_url，以及可选 mime_type/sha256/byte_size。

公共提示词优化的参考素材项严格只传 role、assetId，不要附加 mediaType；生成接口 的参考素材按服务合同传 role、assetId，mediaType 仅在该生成合同要求或支持时传入。 不要传平台内部路径或存储 ID。部署模式使用应用受控 readUrl 及完整性元数据，应用不把 它返回浏览器或写入普通日志。

示例（部署模式）：

job = runtime.create_job(
    {
        "clientTaskId": "story-42",
        "model": "model-alias",
        "serviceKind": "image_generation",
        "params": {"prompt": "一只橘猫"},
        "mediaSpec": {"generationMode": "t2i", "imageCount": 1},
    },
    idempotency_key="story-42",
)
recovery_ref = job.recovery_ref
latest = runtime.get_job(job.job_id, recovery_ref=recovery_ref)
if latest.status == "succeeded":
    for item in latest.results:
        downloaded = runtime.download_result_with_metadata(
            latest.job_id, item.result_id, recovery_ref=recovery_ref
        )
        assert downloaded.byte_size == len(downloaded.content)

不要用预计生成时长代替客户端超时；媒体服务返回的实际能力和超时合同优先。完整媒体 参数与素材限制见媒体完整调用链页面。

AppGrantClient：应用间授权

AppGrantClient(
    *, app_key: str,
    base_url: str | None = None,
    sa_token: str | None = None,
    sa_token_file: str | None = None,
    access_token: str | None = None,
    app_id: str | None = None,
    app_version_id: str | None = None,
    developer_base_url: str | None = None,
    client: httpx.Client | None = None,
    timeout: float | httpx.Timeout = DEFAULT_TIMEOUT,
    limits: httpx.Limits = DEFAULT_LIMITS,
    token_refresh_interval_seconds: float = 1.0,
)

request 接受 Mapping 或规范化 bytes。canonical_request_digest(payload) 对 Mapping 使用 UTF-8、键排序、无空白 JSON，对 bytes 按原字节计算，返回 sha256:<64 位小写十六进制>。AsyncAppGrantClient 提供同名方法，网络方法和 close() 需要 await。

双本地时 resolve、issue_launch 与 RuntimeClient.exchange 自动使用公开调试接口，提供方只用自己的 CLI 凭据。receiverRef 是可分享的配对编号，不是 token；不要分享客户端凭据。正常由 CLI 负责登记和退出时撤销。不同主体、错误目标/能力/摘要或失效配对为受控404，同一幂等键不同内容为409，平台不可用为503；SDK 统一抛 RuntimeAPIError。接收许可最长一小时，grant 最长两分钟，启动码最长一分钟且仅一次，均受源授权期限约束。

AppCenterClient：开发者控制面

该客户端只操作开发者自己的版本，不提供管理员审核或上线能力。

AppCenterClient(
    *, base_url: str | None = None, token: str | None = None,
    tenant_id: str | None = None, client: httpx.Client | None = None,
    upload_timeout_seconds: float | None = None,
)

公开方法为：upload_bundle(app_id, bundle)、upload_source(app_id, version_id, source)、 create_version(app_id, submission, *, source_path=None)、 request(app_id, version_id, kind, reason="")、 upload_image_archive(app_id, version_id, path, *, component, display_version, idempotency_key, archive_type="docker-archive")、 upload_image_archive_for_app(app_id, path, *, component, display_version, idempotency_key, archive_type="docker-archive")、 submit(app_id, version_id) 和 close()；除 close() 外返回 dict[str, Any]。

镜像上传会计算 SHA-256、创建会话、向短期 URL 上传分片并确认完成；默认上传超时 600 秒、连接超时 20 秒。create_version 若提供 source_path，会在发 HTTP 前校验源文件与 待提交 AppSubmission 的规范化 JSON 完全一致。

发布校验相关类型/函数位于 jusp_sdk.appstore：AppProject.from_file、 AppBundle.from_file、SourceReceipt.from_file/to_contract、AppSubmission.to_contract、 validate_project、validate_bundle、validate_submission。根包导出的是常用 Runtime 类型和 AppCenterClient；完整发布辅助类型从 jusp_sdk.appstore 导入。

类型速查

身份、授权和 Job 的 wire DTO 会拒绝未知或缺失的核心字段；目录和结果 DTO 对部分扩展字段 采用默认值/过滤，不能据此假设所有 from_contract 都是完全严格 schema 校验。