Workflows

The WorkflowSpec builder + the multi-role primitives.

WorkflowSpec

from convilyn_sdk import WorkflowSpec

workflow = (
    WorkflowSpec("text_analyzer", name="Text Analyzer")
    .with_description("Extract structure + summary from documents")
    .with_input(types=["document"], formats=["pdf", "txt"])
    .with_output(format="json", additional={"type": "analysis_result"})
    .add_phase("Parse", "Extract text using `analyzer__parse`.")
    .add_phase("Summarize", "Summarize using `analyzer__summarize`.")
)

spec_json = workflow.compile()
workflow.save("workflow.spec.json")

Every with_* / add_* / use_* method returns a new WorkflowSpec — the original is never mutated. This makes mid-stream forking safe and lines up with the immutability convention in the consumer SDK.

Constructor

WorkflowSpec(spec_id: str, *, name: str | None = None)

Metadata methods

I/O contract

workflow = workflow.with_input(types=["document"], formats=["pdf", "txt"], language="en")
workflow = workflow.with_output(format="json", additional={"type": "analysis_result"})

• with_input(types, formats, language=None) — declared input shape; the platform validates uploads against this before invoking the workflow
• with_output(format, **additional) — declared output shape; arbitrary additional keys round-trip into the wire JSON without SDK awareness (OCP: a server-side schema extension does not force an SDK bump)

Wiring tools

workflow = workflow.use_tools("analyzer__parse", "analyzer__summarize")
# or
workflow = workflow.use_servers("analyzer")            # all tools from one server
# or
workflow = workflow.from_server(my_server)             # bind a live ToolServer instance

from_server is the most common shape during local dev: it copies the server's tool refs into the workflow + records the dependency in the manifest. use_tools / use_servers are explicit, useful when wiring an already-deployed server by name.

Agent runtime

workflow = workflow.with_agent_config(
    max_iterations=20,
    temperature=0.3,
    model="claude-sonnet-4-6",
)

with_progress_milestones({"Parse": 30, "Summarize": 80}) lets you control which phase boundaries emit progress events.

Phases + slots + preflight

workflow = (
    workflow
    .add_phase("Parse", "Extract text using `analyzer__parse`.")
    .add_slot("topic", description="Topic the document covers", required=True)
    .add_preflight_rule(
        name="document_present",
        when="$.input.document",
        message="Upload a PDF or TXT document first.",
    )
)

• add_phase(name, description) — phases run in declaration order; description is the LLM-visible instruction
• add_slot(name, ...) — declares a slot the workflow needs the user to fill before it runs
• add_preflight_rule(name, when, message, ...) — block workflow start until a JSON-Path-style predicate holds

Locale

workflow = workflow.with_locale_policy(
    default="en",
    supported=["en", "zh-TW", "zh-CN", "ja", "ko"],
    fallback="en",
)

Multi-role workflows

from convilyn_sdk import AgentRole, RoleConfig

workflow = workflow.with_multi_role(
    roles=[
        RoleConfig(role="extractor", tool_allowlist=["analyzer__parse"]),
        RoleConfig(role="summarizer", tool_allowlist=["analyzer__summarize"]),
    ],
    coordinator_role="extractor",
)

See the Multi-role + checkpoints section below for the full surface.

Composing policies

from convilyn_sdk import RetryPolicy, TimeoutPolicy

workflow = workflow.with_policy(
    RetryPolicy(rules=[FailureRule(error_code="UPSTREAM_TIMEOUT", max_retries=3)]),
    TimeoutPolicy(step_seconds=120),
)

See Policies for the full catalogue and the wire-shape Config siblings.

Compile + load

spec_json = workflow.compile()                # dict[str, Any] — the wire payload
workflow.save("workflow.spec.json")           # writes JSON to disk
loaded = WorkflowSpec.load("workflow.spec.json")

compile() runs validation (see workflow_validator) before returning; load round-trips a compiled JSON back into a builder you can keep mutating.

AgentRole

A structural Protocol — any frozen dataclass (or any object with the right attributes) is an AgentRole:

from convilyn_sdk import AgentRole
from dataclasses import dataclass

@dataclass(frozen=True)
class MyRole:
    role: str
    tool_allowlist: list[str] | None = None

assert isinstance(MyRole(role="extractor"), AgentRole)

Required:

• role: str — canonical identifier the workflow uses to reference the role
• tool_allowlist: list[str] | None — optional per-role tool boundary; None means the role inherits whatever tools the workflow declared

Pass either a bare str (the role identifier only) or any AgentRole to with_multi_role(roles=[...]).

The v1.x alias Specialist resolves to AgentRole. Direct imports of convilyn_sdk.specialist still emit a DeprecationWarning; the top-level Specialist = AgentRole alias stays silent so existing callers do not have to migrate immediately.

Multi-role + checkpoints

MultiRoleConfig

from convilyn_sdk import MultiRoleConfig, RoleConfig

MultiRoleConfig(
    roles=[
        RoleConfig(role="extractor", tool_allowlist=["analyzer__parse"]),
        RoleConfig(role="summarizer"),
    ],
    coordinator_role="extractor",
)

Wraps the multi-agent block on the wire. extra="allow" is set on every model — a server that adds a new optional field does not force an SDK bump.

RoleConfig

RoleConfig(
    role="extractor",                          # required
    tool_allowlist=["analyzer__parse"],         # optional
    description=None,                           # optional
    autonomy=AutonomyLevel.GUIDED,              # optional
)

SpecialistConfigModel is a v1.x deprecation alias of RoleConfig; use RoleConfig in new code.

CheckpointConfig

Checkpoints let a long-running workflow resume after a controlled pause. Add one with WorkflowSpec.with_resume_boundary(name, after_phase=...) or with_checkpoint(name, config):

from convilyn_sdk import CheckpointConfig

workflow = workflow.with_resume_boundary(
    name="after_parse",
    after_phase="Parse",
)

# Or with an explicit config:
workflow = workflow.with_checkpoint(
    "after_parse",
    CheckpointConfig(after_phase="Parse"),
)

The compiled spec ships a checkpoints dict — each entry tells the platform when to snapshot state, and what to do if the user takes too long to respond.

Versioning

Every advanced-types model sets extra="allow". Add a new optional field on the wire and the SDK round-trips it transparently through compile() → model_dump(); consumers do not have to bump for additive changes.