Skiva Python SDK

English · 简体中文

Async Python SDK for Skiva's /distribution/v1 protocol. Python 3.11+. Package name: skiva-sdk; import: skiva_sdk. Version: 0.3.0.

GitHub repository · Skiva server

Install

Install from PyPI:

python -m pip install skiva-sdk==0.3.0
# Optional framework adapter, tested against exactly Pydantic AI 2.59.0:
python -m pip install 'skiva-sdk[pydantic-ai]==0.3.0'

Use a client credential assigned to an enabled Agent Profile, not an admin/API session token. Use the server origin (or reverse-proxy prefix), without /distribution/v1. Use HTTPS outside a trusted local environment.

export SKIVA_SERVER_URL=http://localhost:8080
export SKIVA_CLIENT_TOKEN=your-client-credential

The profile must have a published snapshot and authorized Skills. SDK dynamic mode respects server-side profile pins; it does not bypass them.

Discovery and loading

import asyncio
from skiva_sdk import SkivaClient

async def main():
    async with SkivaClient.from_env() as client:
        async with client.bind() as skills:
            catalog = await skills.catalog()
            print(catalog.render())
            if catalog.skills:
                content = await skills.load_skill(
                    catalog.skills[0].name,
                    expected_snapshot_id=catalog.ref.snapshot_id,
                    expected_catalog_digest=catalog.catalog_digest,
                )
                print(content.render())

asyncio.run(main())

Catalog contains a SnapshotRef, immutable Skill metadata and catalog_digest. SkillContent contains the actual ref, catalog digest, manifest-relative path, file digest and text. Rendered envelopes escape XML markup. Names/descriptions still originate from published Skills: treat publishers as a trust boundary, not as trusted system-policy authors simply because text is escaped.

The discovery prompt tells the model to compare snapshot IDs for equality, not chronological order, and to reload content whose version differs. Rollback is handled the same way as forward updates. catalog_digest also changes when the authorized file set changes within one snapshot. This is a model-facing fallback, not a guarantee that old instructions are forgotten.

Choose an interface

Use these two primary interfaces. bind(consistency="pinned") and bind(source=ref) remain lower-level compatibility options. client.task(id) is just an alias for snapshot(id), not another mode.

Fixed and dynamic policies

async with client.bind(consistency="dynamic") as skills:
    await skills.refresh()  # explicit reconciliation; raises on failure
    async with skills.snapshot() as batch:
        body = await batch.read_text("review/SKILL.md")
        rules = await batch.read_text("review/references/rules.md")
        original_ref = batch.ref

• Opt-in pinned: one snapshot and filtered manifest for the binding. Future channel changes do not change it.
• Default dynamic: reads/discovery reconcile on access after refresh_interval (default 0 seconds: every access). refresh() forces reconciliation. No background daemon or SSE subscription.
• An expected-version load forces dynamic reconciliation even before the interval expires. Snapshot or catalog mismatch raises VersionMismatch; no silent substitution.
• A fully verified candidate replaces the active view atomically. Each operation captures one view; different operations can span versions. snapshot() captures one view for related operations, even while the enclosing dynamic session updates.
• Refresh failure aborts that operation and populates last_refresh_error; it never silently reports stale discovery as fresh. Later successful refresh clears the error. transition_count counts successful view changes.
• The client default can be overridden per binding. An explicit source=SnapshotRef selects pinned mode when consistency is omitted; explicitly combining it with dynamic is rejected. Reuse a ref for an explicit retry, subject to current server grants/permissions.
• view.ref in dynamic mode is a point-in-time observation. Use returned SkillContent.ref or batch.ref for provenance.

Task/attempt IDs are optional correlation fields and auto-generated if absent. Duplicate active IDs are rejected within one client. There is no durable attempt store, automatic crash resume or cross-process exactly-once guarantee. Persist refs in your application's own storage for explicit pinned retries. A ref does not grant access and cannot bypass a new credential's permissions.

Named snapshot bindings

# Default: check the current authorized target on each access.
async with client.bind() as skills:
    catalog = await skills.catalog()

# First entry pins the current snapshot; re-entry reuses it.
async with client.snapshot("review-42") as skills:
    catalog = await skills.catalog()

# Explicitly replace the binding with the latest authorized snapshot.
# This new view stays fixed throughout this block, too.
async with client.snapshot("review-42", retry_policy="latest") as skills:
    catalog = await skills.catalog()

client.forget_snapshot("review-42")

client.snapshot(id) is the primary fixed-view API. The string is an application binding ID, not a server snapshot UUID. client.task(id) is an alias only. Without an ID, client.snapshot() captures an anonymous fixed view. To select an actual snapshot reference, use client.snapshot(source=ref).

Bindings survive application exceptions and context exit, so retrying the same ID reuses the original snapshot by default. retry_policy="latest" replaces the recorded binding only after successful preparation; it does not enable dynamic changes inside the block. Current permissions are always checked. Denied or changed views fail instead of silently moving to latest.

Without a binding_store, the mapping lives only in this client instance. Reuse a long-lived client across requests; separate clients, processes and restarts do not share bindings. Persist SnapshotRef in application storage and pass source=ref to recover across restarts. Concurrent contexts with the same ID raise SnapshotBusy; share the already opened view for parallel reads. The default capacity is 1,024 IDs (max_snapshot_bindings); call forget_snapshot(id) after completing work. Active bindings cannot be forgotten. Forgetting removes only the local mapping, not server snapshots or grants. There is no implicit eviction or expiry that could silently change a retry's version.

Optional durable bindings

from skiva_sdk import SkivaClient, SQLiteBindingStore

store = SQLiteBindingStore("bindings.sqlite", namespace="my-app/review-profile")
async with SkivaClient.from_env(binding_store=store) as client:
    async with client.snapshot("review-42") as skills:
        catalog = await skills.catalog()
        print(catalog.render())
    # A later process using the same database/namespace recovers this binding.
    # Explicit cleanup after the application no longer needs retries:
    # client.forget_snapshot("review-42")

The default store is client-local memory. Opt into SQLiteBindingStore to recover IDs across instances, processes and restarts, always subject to online authorization. Use an application/Profile namespace; server URLs are separated automatically. Place the database in a protected local directory, not a network filesystem. It contains references, not credentials or Skill bodies.

Publication uses atomic compare-and-set: a competing change during preparation raises SnapshotBusy instead of silently overwriting it. One client rejects overlapping contexts with the same ID; separate clients/processes may use their already fixed views concurrently. This is not a distributed execution lock. latest and deletion never mutate existing views; coordinate cleanup after all relevant executions finish. Records never expire implicitly; capacity applies per store namespace. Custom stores implement the atomic get/put/delete methods of BindingStore; methods should be fast local metadata operations.

Manifest ETag caching is bounded to 16 entries and a total of max_manifest_bytes. Every use still contacts the server for authorization; outages and denials never fall back to cached authorization. Fixed snapshots are not offline mode.

Retry policy is separate from update policy

Version 0.2.0 changes the client default from pinned to dynamic. Existing applications that need fixed ordinary bindings must pass consistency="pinned". Retry behavior is independently configured with retry_policy="reuse_snapshot" (default) or "latest", on the client or on client.retry().

async with client.bind(task_id="review-42") as skills:
    catalog = await skills.catalog()
    # Run your application/agent with this session.

checkpoint = skills.retry_checkpoint

# Default: the checkpoint snapshot stays fixed for the whole retry.
async with client.retry(checkpoint) as retry:
    print((await retry.catalog()).render())

# Opt in: resolve the current authorized target and follow the client policy.
async with client.retry(checkpoint, retry_policy="latest") as retry:
    print((await retry.catalog()).render())

retry_checkpoint is an immutable RetryCheckpoint(task_id, attempt_id, snapshot) containing the session's last fully published view, not necessarily its initial view. It is available after the binding exits, including when application code raises. Capture and keep it before further updates if you want that specific view. A failed/cancelled refresh does not advance it. A dynamic attempt may have used several versions: one checkpoint does not reproduce its full sequence of reads or model history. For an exact returned content/batch version, construct a checkpoint with that SkillContent.ref or SnapshotView.ref and the session's IDs.

client.retry(checkpoint) keeps the task ID, generates a new attempt ID, revalidates current permissions and fixes the checkpoint snapshot throughout the retry. It never falls back to latest on denied access or a changed manifest. retry_policy="latest" resolves the current authorized Profile target (including server pins) and then uses the client's consistency policy, dynamic by default. Each retry can produce its own checkpoint for another retry. The checkpoint records no secrets and grants no access; applications can persist it with dataclasses.asdict and reconstruct its nested SnapshotRef when loading.

Reusing a task ID with bind() does not infer a retry. The SDK does not automatically rerun business actions or retry framework tools; wrap the retry boundary you own. For an application failure, read skills.retry_checkpoint in the exception handler after bind exits, then explicitly call client.retry. A failure before the initial binding is prepared has no usable checkpoint. Choosing latest reloads SDK content, but does not erase historical messages; decide whether to start a fresh model run/context in the application.

Pydantic AI integration

from pydantic_ai import Agent
from pydantic_ai.capabilities import Caching
from skiva_sdk import SnapshotView
from skiva_sdk.integrations.pydantic_ai import SkivaSkills

agent = Agent(
    model,  # Supply your configured Pydantic AI model/provider.
    name="skiva-agent",
    deps_type=SnapshotView,
    capabilities=[Caching(), SkivaSkills()],
)

async with agent:
    async with client.snapshot("review-42") as skills:
        result = await agent.run("Review this change", deps=skills)

SkivaSkills provides a named, dynamic discovery instruction part with rewrite semantics, evaluated before every model request. It explicitly refreshes the session at each request, bypassing the normal polling interval. It also registers load_skill and read_skill_file, requiring both expected version fields. Version mismatch/not-found become structured tool results; authorization/network failures abort rather than encourage stale fallback. It does not rewrite historical tool returns or erase derived conclusions. Provider wire placement of instructions depends on Pydantic AI and the provider.

SkivaSkills accepts both SkillSession and SnapshotView. The example uses a fixed named view; for live updates use deps_type=SkillSession and client.bind(). A fixed view rechecks authorization without changing versions.

Keep business dependencies with SkivaSkills(get_session=lambda deps: deps.skills). Each run must receive its own session; the capability has no mutable run state. Attach one SkivaSkills capability per Agent (tool names are fixed). This is a dedicated discovery/tool integration, not a replacement for the native Harness Skills workspace backend. Shell/direct filesystem access is not intercepted. Configure application tracing in the host application; the SDK exports no prompts/tokens to an external telemetry service.

An active binding owns both discovery and reads. Bindings follow updates by default. An explicitly pinned binding, including a reuse-snapshot retry, intentionally advertises its fixed catalog. Instructions are not modified while a model request is in flight. Header changes can affect prompt caching.

Real files when necessary

async with skills.snapshot() as batch:
    async with batch.materialize() as directory:
        # Give this path to a framework or awaited subprocess.
        # The child process must finish before leaving this block.
        print(directory)

This exports a complete private copy into a temporary directory, preserving read/execute modes. It does not create a FUSE mount or writable hardlinks to shared objects. It rejects paths incompatible with supported platforms, including case-folding collisions. Cleanup occurs on exit. File modes are not sandbox isolation against the owning process. Direct reads of exported files bypass SDK authorization; revocation cannot recall exported bytes. New snapshots require new exports, never rewriting an old directory.

Validation, authorization and limits

Every snapshot is fully downloaded and validated before publication: Go-compatible manifest canonicalization, SHA-256, sizes, paths and modes. Cache is in memory and private to each view/client; no shared disk/offline cache. Default limits: 10,000 files, 8 MiB per file/manifest, 64 MiB per snapshot, 256 KiB per SKILL.md parsing input. Concurrent/retained views multiply memory use; applications should limit concurrency and release references.

Every public read/catalog/export rechecks the explicit snapshot manifest online, even for cached bytes. Changed permissions invalidate a captured view; explicit credential revocation is sticky for that client instance. Network failure blocks the operation. No offline mode, transparent network retries, or model-task retries. An authorization race after the final check and previously delivered content cannot be undone. HTTP redirects are not followed; bearer tokens are not included in SDK error messages. Raw transport instrumentation can still expose credentials; configure logging accordingly. Environment proxy/CA settings are ignored by default (trust_env=False); opt in explicitly with trust_env=True if needed, and install httpx[socks] for SOCKS proxies. A custom HTTPX transport can supply application TLS/proxy settings.

Useful errors: VersionMismatch, ManifestChanged, AuthorizationDenied, IntegrityError, ProtocolError, LimitExceeded, SkillNotFound, TransportError, ClosedError, SnapshotBusy.

Development and verification

uv sync --extra dev --extra pydantic-ai
uv run pytest -q
uv run ruff check src tests examples
uv run python -m build
uv run python examples/offline_demo.py

Tests use a controlled distribution fixture and Pydantic's actual FunctionModel agent loop, without external model calls. They cover updates/rollback, same-snapshot permission changes, concurrent reads/runs, old-content provenance, corruption, export isolation, and Go-generated manifest fixtures. The offline demo requires no server or credentials. It is not a substitute for deployment acceptance against your Skiva server and selected model provider.

See examples/ and CHANGELOG.md. Official integration references: Pydantic capabilities, Agent Skills implementation guide.

Detailed scenarios and recorded results: Testing guide.