chaojiyuyan-sdk 0.4.0

Official Python SDK for the chaojiyuyan capability platform — caller-side (device-flow login + capability calls) + worker-side (pull-bid-submit loop) + `chaojiyuyan` CLI + stdio MCP server

chaojiyuyan-sdk

Official Python SDK for the hebing capability platform.

Lets an external developer's agent reach chaojiyuyan capabilities in a few lines — instead of hand-rolling HTTP calls, signing headers, and writing a device-flow polling loop. Ported from the original nexus_sdk, re-targeted at hebing's kernel-job API.

Install

pip install chaojiyuyan-sdk

Or from this repo:

cd sdk && pip install -e .

The only runtime dependency is httpx.

Quick start

from chaojiyuyan_sdk import NexusClient

client = NexusClient.login_device_flow(client_name="my-agent")

job = client.call("media.image.campaign", {"prompt": "a red fox in snow"})
print(job["id"], job["status"])

NexusClient is the primary class. HebingClient is an alias for the same class if you prefer the repo name.

Three ways to authenticate

hebing agent keys look like nxt_ag_.... Pick whichever fits your runtime.

1. Device flow (recommended)

Self-service browser login — no key copy-paste, no key in your shell history. The SDK prints a short code and a URL; you approve in a browser; the SDK polls until approved and saves the key to ~/.chaojiyuyan/credentials.

from chaojiyuyan_sdk import NexusClient

client = NexusClient.login_device_flow(client_name="my-agent")
# prints:
#   Connect to hebing
#   ----------------------------------------
#   1) Open: https://pic.chaojiyuyan.com/activate?code=WXYZ-1234
#   2) Enter code: WXYZ-1234
#
#   Waiting for approval in your browser...

After a successful login the next run can just use from_env() — the saved credentials file is the fallback source.

2. Environment variable

For CI, containers, and any non-interactive runtime.

export CHAOJIYUYAN_API_KEY=nxt_ag_xxxxxxxx
# optional, defaults to production:
export HEBING_BASE_URL=https://pic.chaojiyuyan.com/api/v1

from chaojiyuyan_sdk import NexusClient

client = NexusClient.from_env()

from_env() resolution order: CHAOJIYUYAN_API_KEY (or legacy NEXUS_API_KEY) env var → then the ~/.chaojiyuyan/credentials file.

3. Direct key

When you already hold the key string (e.g. injected by a secret manager).

from chaojiyuyan_sdk import NexusClient

client = NexusClient(api_key="nxt_ag_xxxxxxxx")

Calling a capability

from chaojiyuyan_sdk import NexusClient

with NexusClient.from_env() as client:
    # Browse the catalog.
    for cap in client.capabilities():
        print(cap["capability_key"], "-", cap["name"])

    # Run a capability — returns the created kernel job.
    job = client.call(
        "media.image.campaign",
        {"prompt": "a red fox in snow"},
        idempotency_key="my-unique-request-1",  # optional dedupe
    )
    print("job", job["id"], "status", job["status"])

call() maps to POST /api/v1/kernel-jobs. A fresh job comes back at HTTP 201; an idempotent replay (same idempotency_key, same caller) comes back at HTTP 200 with the original job — the SDK returns the job either way.

Tracking a job to completion

final = client.wait(job["id"], timeout=300)
print(final["status"])  # succeeded / failed / cancelled / expired

wait() polls get_job() until the job reaches a terminal status. A terminal job may still be a failure — always check status.

Agent-key job polling

The job-read endpoints — GET /api/v1/kernel-jobs, GET /api/v1/kernel-jobs/{id}, GET /api/v1/kernel-jobs/{id}/events, and GET /api/v1/capability-specs — accept an agent key (nxt_ag_*) as well as an owner JWT. An agent key obtained via the device flow can therefore call() a capability and poll it back with get_job() / wait(), plus list its own jobs and read the capability catalog with capabilities().

Reads are scoped to the calling agent: get_job() / wait() resolve only jobs the agent created, and list / events behave the same way. Asking for a job that belongs to a different agent — or that does not exist — raises a HebingAPIError with HTTP 404 (the API never confirms a job exists for some other caller).

A webhook / callback is still a fine choice when you want the terminal result pushed to you instead of polling for it.

Error handling

from chaojiyuyan_sdk import NexusClient, HebingAPIError, JobTimeout

try:
    job = client.call("does.not.exist", {})
except HebingAPIError as e:
    print(e.status_code, e.code, e.detail)  # 400 unknown_capability ...

Exception tree (all rooted at HebingError):

Exception	Raised when
HebingAPIError	the API returned a non-2xx response (status_code, code)
DeviceFlowError	the device flow was denied or the code expired (410)
DeviceFlowTimeout	device-flow approval did not arrive within the timeout
JobTimeout	wait() ran out of time before the job settled

Network-level failures (connection refused, DNS, TLS) are not re-wrapped — they bubble up as httpx.HTTPError.

API reference

Method	Endpoint
NexusClient(api_key=...)	—
NexusClient.from_env()	—
NexusClient.login_device_flow()	POST /oauth/device/authorize + POST /oauth/device/token
.call(capability_key, input_payload)	POST /kernel-jobs
.get_job(job_id)	GET /kernel-jobs/{id}
.wait(job_id)	polls GET /kernel-jobs/{id}
.capabilities()	GET /capability-specs

Default base URL: https://pic.chaojiyuyan.com/api/v1.

Worker — accept jobs from hebing

The same SDK ships a worker half. An external agent can plug its capability into hebing's bidding pool with eight lines:

from chaojiyuyan_sdk import Worker

worker = Worker.from_env()  # or Worker(api_key="nxt_ag_...")

@worker.handler("test.echo.v1")
def render(task):
    return {"echo": task.input.get("text", "")}

worker.run()

Worker.run() does six things for you:

1. Auto-register handlers via POST /api/v1/kernel-agents/me/capabilities.
2. Long-poll GET /api/v1/kernel-jobs/available for jobs in your handler set.
3. Bid FCFS via POST /api/v1/kernel-jobs/{id}/bid (first request wins — no bid_credits, hebing PR1 is sync FCFS per ADR-051 §3.3-D).
4. Hold the lease with a background thread calling POST /api/v1/kernel-jobs/{id}/heartbeat every 10s (lease is 30s).
5. Dispatch your handler with a TaskContext carrying job_id, capability_key, input_envelope, max_budget_nc, task_run_id, lease_expires_at.
6. Submit the return value to POST /api/v1/kernel-jobs/{id}/submit. Settle is same-transaction inside the kernel — no separate callback.

The bearer key flows on Authorization: Bearer nxt_ag_* (ADR-056). The worker endpoints require an agent key that has registered at least one capability handler — otherwise the API returns 403 worker_not_enabled and the SDK raises WorkerNotEnabled.

Catchable worker exceptions

All rooted at WorkerError (itself a HebingError):

Exception	HTTP / status	Cause
WorkerNotEnabled	403	No handler registered yet — call register_handlers().
CapabilityNotRegistered	403	This worker never registered the job's capability.
BidRejected	200 accepted=False, 410	Lost the race, bid window expired, or attempts exhausted.
TaskRunNotFound	404	Heartbeat/submit landed after re-route.
LeaseExpired	410	Heartbeat or submit landed past the 30s lease.
SubmitValidationFailed	422	Output failed the capability's validation; retries left.

Test-friendly knobs

worker.run(once=True, heartbeat_interval=1000)  # one polling cycle, no heartbeat

Worker(http_client=httpx.Client(transport=httpx.MockTransport(...))) lets tests replace the entire HTTP stack with an offline mock.

Webhook verification

When the kernel ships a settlement webhook (X-Kernel-Signature header, sha256=<hex> body), verify it inline:

from chaojiyuyan_sdk import verify_webhook_signature

if not verify_webhook_signature(
    request.headers["X-Kernel-Signature"],
    request.body,
    current_secret=os.environ["HEBING_WEBHOOK_SECRET"],
):
    return 401

Supports rotated-secret grace periods via previous_secret + previous_secret_expires_at.

CLI — create KEY, then connect this machine

Use the web console to create a nxt_ag_* Agent KEY, then run the PATH-safe module command below. It saves the KEY to ~/.chaojiyuyan/credentials and wires Claude Code's MCP config:

python3 -m pip install --user 'chaojiyuyan-sdk[mcp]'
python3 -m chaojiyuyan_sdk.cli connect --api-key 'nxt_ag_...' --yes

After that, browse and call capabilities from the shell:

python3 -m chaojiyuyan_sdk.cli capabilities
python3 -m chaojiyuyan_sdk.cli call media.echo.v1 --input '{"text":"hi"}' --wait
python3 -m chaojiyuyan_sdk.cli status <job_id>

Or run a worker (PR3 ships an echo-stub for the demo — real workers write their own handler using chaojiyuyan_sdk.Worker):

python3 -m chaojiyuyan_sdk.cli worker media.echo.v1 --yes

If your Python user scripts directory is already on $PATH, the shorter chaojiyuyan ... console command is equivalent. If you want device-flow login instead of pasting a web KEY, run python3 -m chaojiyuyan_sdk.cli connect --yes.

Full subcommand list

Subcommand	What it does
chaojiyuyan connect	Save a web KEY or run device-flow, then write mcpServers.chaojiyuyan into ~/.claude/mcp.json.
chaojiyuyan disconnect	Clear ~/.chaojiyuyan/credentials + remove mcpServers.chaojiyuyan from mcp.json.
chaojiyuyan status [JOB_ID]	No arg → local profile (masked key + base_url). With arg → get_job(...).
chaojiyuyan capabilities	List the kernel capability catalog.
chaojiyuyan call <cap> ...	Create a kernel job + optionally --wait for terminal status.
chaojiyuyan worker <cap>	PR3 demo: run an echo-stub worker against <cap>. Confirms before starting.
chaojiyuyan mcp-server	Run the stdio MCP server (Claude Code / Desktop subprocess). Needs [mcp] extras.
chaojiyuyan version	Print the SDK version.

mcp.json safety

chaojiyuyan connect writes to ~/.claude/mcp.json by default. It is strict about not clobbering existing config:

• Other mcpServers entries are preserved verbatim — only the chaojiyuyan block is touched.
• Top-level keys outside mcpServers are untouched.
• A timestamped backup (mcp.json.backup-YYYYMMDDHHMMSS) is created in the same directory before the new file is written, so a rollback is always one mv away.
• The file is chmod 600 on POSIX (it holds the API key in env).
• Use --dry-run to see the diff without writing, or --target <path> to point at a different MCP config (e.g. Claude Desktop).

See docs/DESIGN-PR3-mcp-json-writer.md for the full algorithm spec.

MCP server — chaojiyuyan mcp-server

The chaojiyuyan connect command writes mcpServers.chaojiyuyan into your MCP config pointing at chaojiyuyan mcp-server. That subcommand is a stdio MCP server that Claude Code / Desktop spawn as a subprocess to surface hebing tools to the model.

chaojiyuyan-sdk requires Python 3.10+. The MCP packages don't ship with the slim base install — opt in:

pip install 'chaojiyuyan-sdk[mcp]'

After that, chaojiyuyan connect (or chaojiyuyan mcp-server directly) just works.

MCP tools

Tool	Side	What it does
kernel_list_capabilities	caller	Browse the catalog
kernel_create_job	caller	POST /api/v1/kernel-jobs
kernel_get_job	caller	GET /api/v1/kernel-jobs/{id}
kernel_wait_for_job	caller	Poll until terminal (or timeout)
kernel_list_artifacts	caller	Slice job.artifacts from kernel_get_job
kernel_register_capability_handler	worker	POST /kernel-agents/me/capabilities
kernel_claim_job	worker	Bid on one available job and hold its lease
kernel_submit_job	worker	Submit output for a claimed job

kernel_claim_job keeps the worker-pull lease alive with a background heartbeat inside the stdio MCP server. The claim is bounded by max_hold_seconds (default 600) so an abandoned MCP session does not hold a job forever. kernel_submit_job stops the local heartbeat after a successful submit.

Auth

CHAOJIYUYAN_API_KEY (or legacy HEBING_API_KEY / NEXUS_API_KEY) from env — the mcp.json env block that chaojiyuyan connect wrote puts it there. Falls back to ~/.chaojiyuyan/credentials if env is empty. Missing → single stderr line + exit 1, no Python traceback.

Error responses from tools carry stable code strings (unknown_capability, insufficient_balance, job_timeout, bad_arguments, …) so the upstream model can pattern-match without parsing free-form error text.

See docs/DESIGN-PR4-mcp-stdio-server.md for the full algorithm spec.

Manual end-to-end smoke (Claude Code subprocess)

Once everything's wired, verify the full chain with a local install:

cd sdk
python3.10 -m pip install -e '.[mcp]'  # SDK + optional [mcp] extras
chaojiyuyan version                    # confirms `chaojiyuyan` is on $PATH
chaojiyuyan connect --dry-run          # confirms the mcp.json diff renders
chaojiyuyan capabilities               # confirms the API call works

For the Claude Code subprocess path, after a real (non---dry-run) chaojiyuyan connect:

# Launch Claude Code. In the host:
#   Tools → check that `chaojiyuyan` shows up with kernel_* tools listed.
#   Then prompt: "Call hebing kernel_list_capabilities."
# Verify the response includes the catalog.

scripts/smoke-chaojiyuyan-sdk.sh automates the non-interactive portion of the above — see the top of that file for usage.

Development

cd sdk
python -m pytest tests/ -q

Related

• ADR-051 — SDK worker onboarding (ratified 2026-05-25), §3.10 G1 (pip install chaojiyuyan-sdk + chaojiyuyan CLI)
• ADR-054 — MCP worker access, §3.10 G2 (~/.claude/mcp.json auto-write)
• DESIGN-PR1-worker-pull — endpoint signatures + state machine
• PR1 implementation — platform/api/kernel_worker_pull.py
• PR3 design — docs/DESIGN-PR3-mcp-json-writer.md
• PR4 design — docs/DESIGN-PR4-mcp-stdio-server.md

License

MIT