toolvingImplement the three workflows with the Python SDK
The program below is complete and runnable. It teaches the same workflow planner, records filing, and invoice extraction contracts used by the guided demos, probes a novel input, records an authoritative outcome, saves the construction-supporting teaching text, ingests it into a new instance, and retires both instances.
Install and configure
Download and verify SDK 0.12.1 from the releases page, then install its wheel:
python -m pip install ./toolving-0.12.1-py3-none-any.whl
export TOOLVING_API_KEY='your-account-api-key'
export TOOLVING_API_URL='https://toolving.com'
Keep the account key on your server. Do not put it in browser JavaScript, a mobile application, source control, logs, or a public environment file.
Save the following as conceiver_workflows.py and run python conceiver_workflows.py.
import json
import os
from dataclasses import dataclass
from typing import Any, Callable
from uuid import uuid4
from toolving import ConceiverClient
MAX_INPUT_BYTES = 8_000
MAX_ANSWER_BYTES = 16_000
@dataclass(frozen=True)
class Workflow:
name: str
teachings: tuple[tuple[str, Any], ...]
probe: str
authoritative_answer: Any
encode: Callable[[Any], str]
decode: Callable[[str], Any]
def validate_input(value: str) -> str:
if not value or "\n" in value:
raise ValueError("A workflow input must be one non-empty line.")
if len(value.encode("utf-8")) > MAX_INPUT_BYTES:
raise ValueError("A workflow input exceeds 8,000 UTF-8 bytes.")
return value
def canonical_json(value: Any) -> str:
# Conceiver's language API accepts one-line strings. Compact JSON keeps
# each structured answer on that contract.
encoded = json.dumps(value, separators=(",", ":"), sort_keys=True)
if len(encoded.encode("utf-8")) > MAX_ANSWER_BYTES:
raise ValueError("A workflow answer exceeds 16,000 UTF-8 bytes.")
return encoded
def parse_json(value: str) -> Any:
if "\n" in value or len(value.encode("utf-8")) > MAX_ANSWER_BYTES:
raise ValueError("A structured answer violates the local output limit.")
return json.loads(value)
def validate_plan(value: Any) -> list[dict[str, Any]]:
if not isinstance(value, list) or not value:
raise ValueError("An onboarding plan must be a non-empty list.")
for index, call in enumerate(value):
if not isinstance(call, dict) or set(call) != {"tool", "args"}:
raise ValueError(f"Tool call {index} must contain tool and args.")
if not isinstance(call["tool"], str) or not call["tool"]:
raise ValueError(f"Tool call {index} needs a tool name.")
if not isinstance(call["args"], dict):
raise ValueError(f"Tool call {index} args must be an object.")
return value
def encode_plan(value: Any) -> str:
return canonical_json(validate_plan(value))
def decode_plan(value: str) -> list[dict[str, Any]]:
return validate_plan(parse_json(value))
def validate_invoice(value: Any) -> dict[str, str]:
fields = ("invoice", "vendor", "amount", "due")
if not isinstance(value, dict) or set(value) != set(fields):
raise ValueError(f"An invoice must contain exactly: {', '.join(fields)}.")
if any(not isinstance(value[field], str) or not value[field] for field in fields):
raise ValueError("Every invoice field must be a non-empty string.")
return {field: value[field] for field in fields}
def encode_invoice(value: Any) -> str:
return canonical_json(validate_invoice(value))
def decode_invoice(value: str) -> dict[str, str]:
return validate_invoice(parse_json(value))
def encode_text(value: Any) -> str:
if not isinstance(value, str) or not value or "\n" in value:
raise ValueError("The filing destination must be one non-empty line.")
if len(value.encode("utf-8")) > MAX_ANSWER_BYTES:
raise ValueError("A workflow answer exceeds 16,000 UTF-8 bytes.")
return value
def decode_text(value: str) -> str:
return encode_text(value)
def onboarding_plan(user: str, team: str) -> list[dict[str, Any]]:
return [
{"tool": "create_account", "args": {"user": user, "team": team}},
{"tool": "add_to_team", "args": {"user": user, "team": team}},
{"tool": "send_welcome", "args": {"user": user}},
]
WORKFLOWS = (
Workflow(
name="onboarding",
teachings=(
("onboard priya to engineering", onboarding_plan("priya", "engineering")),
("onboard marco to sales", onboarding_plan("marco", "sales")),
("onboard leila to marketing", onboarding_plan("leila", "marketing")),
),
probe="onboard amina to operations",
authoritative_answer=onboarding_plan("amina", "operations"),
encode=encode_plan,
decode=decode_plan,
),
Workflow(
name="filing",
# Vary both category and year. Repeating one category with new dates
# did not provide enough evidence for a category transformation.
teachings=(
("file council minutes 2025", "records/council/2025"),
("file finance minutes 2026", "records/finance/2026"),
("file parks minutes 2027", "records/parks/2027"),
),
probe="file planning minutes 2028",
authoritative_answer="records/planning/2028",
encode=encode_text,
decode=decode_text,
),
Workflow(
name="extraction",
teachings=(
(
"Invoice INV-2201 from Acme for 1200 due 2025-10-01",
{
"invoice": "INV-2201",
"vendor": "Acme",
"amount": "1200",
"due": "2025-10-01",
},
),
(
"Invoice INV-3001 from Northwind for 4400 due 2026-01-31",
{
"invoice": "INV-3001",
"vendor": "Northwind",
"amount": "4400",
"due": "2026-01-31",
},
),
(
"Invoice INV-4010 from Contoso for 975 due 2026-02-28",
{
"invoice": "INV-4010",
"vendor": "Contoso",
"amount": "975",
"due": "2026-02-28",
},
),
),
probe="Invoice INV-5012 from Tailspin for 80 due 2026-05-01",
authoritative_answer={
"invoice": "INV-5012",
"vendor": "Tailspin",
"amount": "80",
"due": "2026-05-01",
},
encode=encode_invoice,
decode=decode_invoice,
),
)
def run_workflow(client: ConceiverClient, workflow: Workflow) -> None:
# Keys are tenant-wide idempotency identities. A fresh run identifier makes
# the script rerunnable while preserving stable keys inside this run.
run_id = uuid4().hex
def key(step: str) -> str:
return f"example-{workflow.name}-{run_id}-{step}"
instance = client.create(key=key("create"))
sequence_id = None
try:
for index, (question, answer) in enumerate(workflow.teachings, start=1):
instance.teach(
validate_input(question),
workflow.encode(answer),
key=key(f"teach-{index}"),
)
prediction = instance.query(
validate_input(workflow.probe),
with_memory=True,
key=key("query"),
)
print(f"{workflow.name}: {prediction.status} ({prediction.confidence})")
if prediction.evidence is not None:
print(prediction.evidence.markdown)
print("supporting episodes:", len(prediction.evidence.episodes or ()))
if prediction.status == "ANSWER" and prediction.text is not None:
proposed_answer = workflow.decode(prediction.text)
print("proposed answer:", proposed_answer)
else:
# Abstention is an expected result. Do not invent or execute an
# answer; obtain one from an authoritative downstream source.
print("Conceiver abstained; authoritative downstream work is required.")
# This sample's value is fixed test data, independent of the prediction.
# In an application, replace it with the actual operator or downstream
# result. Never confirm a prediction merely because Conceiver produced it.
actual_answer = workflow.authoritative_answer
instance.resolve(
prediction.prediction_id,
workflow.encode(actual_answer),
key=key("resolve"),
)
sequence_id = instance.save(key=key("save"))
finally:
retirement = instance.reap(key=key("reap"))
print(
"retired:",
retirement.retired,
"persisted:",
retirement.persisted,
"artifact:",
retirement.artifact,
)
if sequence_id is None:
raise RuntimeError(f"{workflow.name} did not produce a saved sequence")
restored = client.create(sequence_id=sequence_id, key=key("restore"))
try:
replay = restored.query(
validate_input(workflow.probe), key=key("replay-query")
)
if replay.status != "ANSWER" or replay.text is None:
explanation = (
replay.evidence.markdown
if replay.evidence is not None
else "No evidence returned."
)
raise RuntimeError(f"{workflow.name} replay abstained: {explanation}")
print("replayed answer:", workflow.decode(replay.text))
finally:
restored.reap(key=key("replay-reap"))
def main() -> None:
api_key = os.environ["TOOLVING_API_KEY"]
api_url = os.environ.get("TOOLVING_API_URL", "https://toolving.com")
with ConceiverClient(api_url, api_key) as client:
for workflow in WORKFLOWS:
run_workflow(client, workflow)
if __name__ == "__main__":
main()
What to replace in production
The teaching pairs and authoritative_answer values are fixtures. Replace them with outcomes from your operator, database, deterministic business process, or another downstream authority. Keep schemas, capability allowlists, extraction limits, and execution authorization in your application. Conceiver proposes and learns answers; it does not authorize tool calls, move records, or approve invoices.
Always check prediction.status before reading or acting on an answer. UNRESOLVED is an abstention, not an empty success. prediction.evidence.markdown explains the decision. with_memory=True also returns the relevant complete teaching episodes.
The script uses three teachings, one memory-bearing query, one resolution, and one standard replay query per workflow: 23 credits per workflow and 69 credits for all three. Rerunning it spends another 69 credits and adds another set of evidence. These totals follow the current prices of 5 credits for teaching or resolution, 2 for a memory-bearing query, and 1 for a standard query.
save() produces a replayable plain-text training sequence from construction-supporting teaching evidence. reap() releases the live instance. On an account with storage, final retirement can return a verified artifact; without storage it returns persisted=False. The saved sequence remains available for explicit replay in either case.