Python SDK

pip install "lingxigraph[sdk]"

同步客户端

from lingxigraph.sdk import LingxiGraphClient

with LingxiGraphClient(
    "https://agents.example.com",
    token="...",
    timeout=30,
) as client:
    graph = client.graphs.get("production-support")
    assistant = client.assistants.create(
        graph_id=graph["id"],
        graph_version=graph["version"],
        name="support",
    )
    thread = client.threads.create(metadata={"external_id": "ticket-42"})
    run = client.runs.create(
        assistant["id"],
        thread_id=thread["id"],
        input={"request": "reset access", "result": ""},
        max_model_calls=8,
    )

    # Run 创建后即可提交 steering；无论它仍在排队还是已经 running，
    # 成功都只代表已 durably 写入，不代表图已经消费。
    accepted = client.runs.steer(
        run["id"],
        kind="user_input",
        payload={"message": "改成用中文回复"},
        idempotency_key="msg-123",
    )
    print(accepted["id"], accepted["status"])  # "pending"

    for event in client.runs.stream(run["id"]):
        print(event["sequence"], event["kind"])

    completed = client.runs.join(run["id"], timeout=120)
    print(completed["output"])

可用资源方法：

Resource	方法
`graphs`	`list()`、`get(graph_id)`
`assistants`	`create(**values)`、`list()`、`get()`、`update()`、`delete()`
`threads`	`create()`、`list()`、`get()`、`update()`、`state()`、`history()`、`fork()`、`delete()`
`runs`	`create()`、`get()`、`list(thread_id)`、`join()`、`stream()`、`cancel()`、`steer()`、`resume()`
`store`	`batch(operations)`、`search(namespace, ...)`
`schedules`	`create()`、`list()`、`update()`、`delete()`

Redrive 或尚未封装的端点可使用 client.request(method, path, **kwargs)。

`runs.steer()`

client.runs.steer(
    run_id,
    kind="user_input",
    payload={"message": "..."},
    metadata=None,
    idempotency_key="msg-123",
)

kind 默认 "user_input"，可自定义为任意业务字符串；
payload 是图代码通过 runtime.drain_steering() 读回的原始 JSON；
idempotency_key 传入时会同时作为 Idempotency-Key header 发送；同一 key 重复调用返回原事件，不会重复入队；
返回值是 202 响应体（含 id、sequence、status），不代表图已经处理这条输入—— 只有服务端后续出现的 run.steer.consumed 事件才代表真正被消费；
对处于终止态的 run 调用会抛出 LingxiGraphAPIError，code == "run_terminal"；对已被 resume 取代的旧 run_id 调用会抛出 code == "run_superseded"。

异步客户端

from lingxigraph.sdk import AsyncLingxiGraphClient

async with AsyncLingxiGraphClient(base_url, token=token) as client:
    thread = await client.threads.create()
    run = await client.runs.create(
        assistant_id,
        thread_id=thread["id"],
        input={"request": "hello"},
    )

    accepted = await client.runs.steer(
        run["id"],
        kind="user_input",
        payload={"message": "switch to Spanish"},
        idempotency_key="msg-123",
    )
    print(accepted["id"], accepted["status"])

    async for event in client.runs.stream(run["id"]):
        print(event)

异步资源与同步资源一一对应。不要在 async 事件循环中使用同步客户端。

错误处理

from lingxigraph.sdk import LingxiGraphAPIError

try:
    client.runs.get("missing")
except LingxiGraphAPIError as exc:
    print(exc.status_code, exc.code, exc.request_id)
    if exc.retryable:
        retry_with_backoff()

只根据 status_code、稳定 code 与 retryable 分支；detail 用于诊断，不是机器协议。SDK 不会自动重试创建 run；调用方应发送稳定 Idempotency-Key（可通过底层 request），防止网络重试重复入队。

同步客户端

runs.steer()

异步客户端

错误处理

On this page

`runs.steer()`