torivers-automation-creator

name	torivers-automation-creator
description	Build, scaffold, and implement automations for the ToRivers marketplace using the torivers-sdk. Invoke this skill whenever a developer wants to create a new ToRivers automation, scaffold a torivers-sdk project, implement a LangGraph-based workflow, integrate external credentials (Gmail, Google Sheets, and other supported services), write LocalSandbox tests, validate a project, or submit to the marketplace. Trigger on any request involving torivers-sdk, `torivers init`, automation workflows, or building marketplace integrations — even when the user does not say "automation" explicitly. Also invoke for requests mentioning third-party service integrations (Slack, Notion, GitHub, etc.) so the skill can advise whether the service is yet available. Also invoke when modifying or debugging an existing torivers-sdk automation project.
license	MIT
metadata	{"author":"torivers","version":"1.2.0","type":"generator","mode":"collaborative","domain":"development"}
allowed-tools	Bash Read Write Edit Glob Grep Agent WebFetch

ToRivers Automation Creator

A ToRivers automation is a LangGraph-based Python workflow packaged with a marketplace manifest. Users purchase and execute automations; the platform handles billing, credential proxying, progress streaming, and output rendering.

This skill is for third-party marketplace developers. All code uses the public torivers-sdk package exclusively. An automation is a standalone Python project — it lives outside the ToRivers platform source code. Even if you happen to have access to a ToRivers codebase, do not reach into it.

The only modules an automation may import are:

1. torivers_sdk.* — the public SDK
2. The stdlib subset in the capabilities snapshot under allowed_imports.stdlib
3. The third-party packages in the capabilities snapshot under allowed_imports.third_party

If a module is not in one of those three lists, the submission validator will reject it. There is no separate blocklist to memorize — the allowlist is the contract.

---

Capabilities Awareness

The set of supported credentials, approved LLM models, allowed categories, and manifest limits evolves with the platform. Do not memorize them. Look them up at the start of every session in this tier-down order:

1. Live capabilities endpoint — at the start of Phase 1, run:

   # Idempotent — required because some distribution channels (zip downloads,
   # skill packagers) don't preserve the execute bit on shell scripts.
   chmod +x scripts/*.sh
   bash scripts/fetch-capabilities.sh > /tmp/torivers-caps.json
   

   The script tries https://torivers.com/api/public/sdk/capabilities first and falls back to the bundled snapshot if unreachable. The stdout is JSON; parse it and keep the relevant subset in context. Run this once per session, not per question.

2. Bundled snapshot — references/capabilities-snapshot.json is the versioned fallback that ships with the skill. The fetch script reads it automatically when the endpoint is down. You can also read it directly.

3. SDK source via Context7 — if the user's environment has the Context7 MCP server installed, use mcp__context7__resolve-library-id + query-docs for torivers-sdk API signatures. This is enrichment, not a substitute for the capabilities snapshot.

4. Ask the user — only as a last resort, e.g., "the snapshot lists the following as fully usable: gmail, google_sheets. Is the integration you need one of these, or one of the 'Coming Soon' services in the snapshot?"

When reasoning about what's available — services, models, categories, limits — cite the snapshot/endpoint, not your training data. The snapshot is the source of truth; your priors are stale by definition.

---

Phase 1 — Clarify Intent

If modifying an existing automation, read references/modifying-existing.md and start there. Don't re-run the full design questionnaire for in-shape changes.

For a new automation, make sure you've run the capabilities fetch from the Capabilities Awareness section above (/tmp/torivers-caps.json exists). If not, run it now.

Then ask the minimum-viable questions before drafting a design:

1. Purpose — what does this automation do, in one sentence?
2. Trigger — on-demand (standard), webhook, or scheduled?
3. Inputs — what does the user provide? Field names + types.
4. External services — which credentials from the capabilities list?

Deepen only when needed:

• Outputs — text report, downloadable files, structured data, or a combination?
• LLM usage — if AI is involved, what for? (Insights and writing — never calculations.)
• Workflow shape — independent steps that could run in parallel? Conditional branches?
• Pricing — what base_price (credits per execution) fits complexity and value?

Skip questions whose answers are obvious from context. Don't interrogate the user; move forward as soon as the picture is clear.

---

Phase 2 — Design, Confirm, Persist

2.1 — Pick an archetype

Most requests fit one of five patterns. Pick one before drafting the design — it gives you a known workflow shape:

• A. Tabular Analysis → Insights — dataset in, stats + narrative out
• B. Cron Digest — scheduled summary of recent activity
• C. Webhook Enrichment — external event in, enriched data out
• D. Multi-Source Aggregation — parallel fetches → combine → analyze
• E. Outbound Notification — compose message → send via email or outbound HTTP

Full descriptions, state sketches, and pricing ranges are in references/archetypes.md. Read it now if you're not sure which fits.

2.2 — Present the design block

Show the design and get explicit confirmation before writing code:

Automation: <kebab-case-name>
Title:      <Human-readable title>
Archetype:  <A | B | C | D | E>
Type:       standard | webhook | scheduled
Duration:   ~<N> seconds
Price:      <N> credits

Workflow steps:
  1. <node-name>: <what it does>  [parallel] [LLM] [<credential>]
  2. ...

Inputs:
  <field>: <type> — <description>

Outputs:
  <block-type>: <what it shows>

Required credentials: <service names or none>
Optional credentials: <service names or none>

2.3 — Persist the design

After the user confirms, the design lives in two files inside the scaffolded project (you'll create them in Phase 3, immediately after torivers init):

• DESIGN.md — the design block above, expanded with a short rationale per decision (why this archetype, why this pricing, why these credentials). The developer reads this when revisiting the project months later; future sessions read it when asked to modify the automation.
• DECISIONS.md — an append-only log. Starts empty. Every later change appends a dated entry explaining what changed and why.

These are real project documentation, not throwaway notes. Treat them as part of the deliverable. The full convention is in references/modifying-existing.md.

---

Phase 3 — Scaffold

Verify the environment, then create the standalone project directory.

chmod +x scripts/*.sh    # safe to re-run; covers the case where this is the
                         # first script in this session (no-op if already set)
bash scripts/check-setup.sh

Scaffold in a new standalone directory (not inside any existing platform codebase):

torivers init <name> [--template basic|data-pipeline|ai-analysis|webhook-handler]
cd <name>

Template guidance (the list comes from the capabilities snapshot if it differs):

• basic — default for most automations (archetype E)
• data-pipeline — processing files or tabular data (archetype A, D)
• ai-analysis — LLM calls are central (archetype A, B)
• webhook-handler — triggered by incoming webhook events (archetype C)

What torivers init generates:

<name>/
├── automation.yaml        # Manifest — update after scaffolding
├── main.py                # Automation class + build_graph() — all SDK imports
├── state.py               # TypedDict state — extend BaseState only
├── requirements.txt       # Add runtime dependencies here
├── README.md              # Marketplace listing description
└── tests/
    └── test_<name>.py     # LocalSandbox tests go here

Immediately after torivers init, write DESIGN.md and create an empty DECISIONS.md. Then update automation.yaml with the agreed metadata. See references/manifest-fields.md for the complete field reference.

---

Phase 4 — Implement

Work through the four files in order. Use sub-agents for independent work when the project is non-trivial — see § Sub-agent Collaboration below.

4.1 — state.py

Extend BaseState (from torivers_sdk.automation.state) with custom fields. Use the provided reducers to control how concurrent node updates are merged:

from typing import Annotated, Any
from torivers_sdk.automation.state import BaseState, merge_dict, append_list, last_value

class MyAutomationState(BaseState):
    # Use append_list when multiple parallel nodes write to the same list
    results: Annotated[list[dict[str, Any]], append_list]

    # Use merge_dict when multiple parallel nodes update the same dict
    stats: Annotated[dict[str, Any], merge_dict]

    # Use last_value when only one node writes to this field at a time
    report_url: Annotated[str, last_value]
    current_item: Annotated[str, last_value]

BaseState already provides: execution_id, input_data, output_data, context, credentials, current_step, errors. Do not redefine these.

Reducer rules:

• append_list — concatenates lists; safe for parallel writes
• merge_dict — right-side dict wins on key conflicts; safe for parallel writes
• last_value — returns the newer value; only for fields one node writes at a time

4.2 — Node functions

Write each workflow step as a plain Python function (sync or async). Each node receives the full state and returns only the keys it updates.

All SDK imports use torivers_sdk.*:

from torivers_sdk.automation.state import BaseState
from torivers_sdk.tools.llm import LLMClient, LLMMessage
from torivers_sdk.tools.storage import StorageClient
from torivers_sdk.tools.http import HttpClient
from torivers_sdk.automation.output import build_output_data, TextBlock, JsonDataBlock


async def fetch_data_node(state: MyAutomationState) -> dict:
    ctx = state["context"]
    credentials = state["credentials"]
    input_data = state["input_data"]

    ctx.log_progress("Fetching data", "Connecting to Google Sheets...")

    # Credential access — only for services declared in the manifest
    sheets = credentials.get_client("google_sheets")
    rows = await sheets.get_range_values(input_data["spreadsheet_id"], "A:Z")

    ctx.log_success("Data loaded", f"{len(rows)} rows loaded.")
    return {"raw_rows": rows, "current_step": "fetch_data"}


async def analyze_node(state: MyAutomationState) -> dict:
    ctx = state["context"]
    ctx.log_progress("Analyzing", "Computing statistics...")

    # Deterministic analysis — use pandas/numpy, NEVER an LLM
    import pandas as pd
    df = pd.DataFrame(state["raw_rows"][1:], columns=state["raw_rows"][0])
    stats = df.describe().to_dict()

    ctx.log_success("Analysis complete")
    return {"stats": stats}


async def generate_insights_node(state: MyAutomationState) -> dict:
    ctx = state["context"]
    ctx.log_progress("Generating insights", "Interpreting the results...")

    llm = LLMClient.get_default()
    # Pass a compact summary to the LLM — never raw rows, never arithmetic
    summary = format_stats_summary(state["stats"])
    response = await llm.chat([
        LLMMessage(role="system", content="You are a data analyst. Provide 3 key insights."),
        LLMMessage(role="user", content=summary),
    ], model="gpt-4o-mini")

    ctx.log_success("Insights ready")
    return {"insights_text": response.content}


async def finalize_node(state: MyAutomationState) -> dict:
    ctx = state["context"]
    output = build_output_data(
        TextBlock(label="Insights", content=state["insights_text"], format="markdown"),
        JsonDataBlock(label="Statistics", data=state["stats"], display_hint="table"),
    )
    ctx.log_success("Done", "Your report is ready.")
    return {"output_data": output, "current_step": "done"}

Key rules:

• Log ctx.log_progress() before a step, ctx.log_success() or ctx.log_error() after
• LLM receives only compact summaries — never raw data, never does arithmetic
• Use credentials.get_client("service") only for services declared in the manifest
• Return a partial dict — only the keys that changed
• For parallel fan-out, conditional routing, retry policies, and async patterns see references/patterns.md
• For full method signatures on HttpClient, StorageClient, LLMClient, GmailClient, GoogleSheetsClient, and their response/error shapes, see references/sdk-api.md

4.3 — main.py (Automation class + graph)

from langgraph.graph import StateGraph, START, END
from torivers_sdk.automation.base import Automation
from torivers_sdk.automation.metadata import AutomationMetadata

from state import MyAutomationState
# from agents import fetch_data_node, analyze_node, ...  (if split into a separate file)


class MyAutomation(Automation[MyAutomationState]):

    def metadata(self) -> AutomationMetadata:
        return AutomationMetadata(
            name="my-automation",
            title="My Automation",
            version="1.0.0",
            description="What this automation does for the user.",
            base_price=2.0,
            category="productivity",
            tags=["data", "analysis"],
            required_credentials=["google_sheets"],
            input_schema={
                "type": "object",
                "properties": {
                    "spreadsheet_id": {"type": "string", "description": "Google Sheets ID"},
                },
                "required": ["spreadsheet_id"],
            },
        )

    def get_state_class(self):
        return MyAutomationState

    def get_required_credentials(self) -> list[str]:
        return ["google_sheets"]

    def build_graph(self) -> StateGraph:
        workflow = StateGraph(MyAutomationState)

        workflow.add_node("fetch_data", fetch_data_node)
        workflow.add_node("analyze", analyze_node)
        workflow.add_node("generate_insights", generate_insights_node)
        workflow.add_node("finalize", finalize_node)

        workflow.add_edge(START, "fetch_data")
        workflow.add_edge("fetch_data", "analyze")
        workflow.add_edge("analyze", "generate_insights")
        workflow.add_edge("generate_insights", "finalize")
        workflow.add_edge("finalize", END)

        return workflow  # Never call .compile() — the runtime does it


# Required: the runtime discovers the automation via this module-level export
automation = MyAutomation()

Rules:

• Always export automation = MyAutomation() at module level
• Never call workflow.compile() inside build_graph()
• Declare credentials in both get_required_credentials() AND AutomationMetadata
• For scheduled automations add schedule_config to metadata
• For automations that accept incoming HTTP triggers, add a webhooks: block (no separate webhook automation type — see references/manifest-fields.md)

4.4 — automation.yaml

torivers init generates this file from your scaffold answers. Review and update it to match the agreed design, then move to Phase 4.5.

4.5 — Verify before tests

After the four files are written, run torivers validate immediately. Don't wait for Phase 6. Fix every error and warning before writing tests — a manifest error will block the test runner and waste your time:

torivers validate

If validation surfaces a missing required field, fix the manifest. If it flags a disallowed import, fix the source. If it complains about pricing or schema mismatches between AutomationMetadata and automation.yaml, reconcile them.

The goal of this gate is: never enter Phase 5 with a project that can't load.

---

Phase 5 — Test

Write tests in tests/test_<name>.py. Use LocalSandbox for end-to-end integration tests and mock clients for unit tests. All test utilities are in torivers_sdk.testing.*:

import pytest
from torivers_sdk.testing import LocalSandbox
from torivers_sdk.context.mocks import MockCredentialProxy
from torivers_sdk.testing.assertions import assert_success, assert_output_contains

@pytest.mark.asyncio
async def test_happy_path():
    from main import automation

    creds = MockCredentialProxy.with_google_sheets()
    creds.get_client("google_sheets").set_mock_data(
        "sheet-id", "Sheet1",
        [["Name", "Revenue"], ["Alpha", 1000], ["Beta", 2500]],
    )

    sandbox = LocalSandbox()
    sandbox.set_credentials(creds)

    result = await sandbox.run(automation, {"spreadsheet_id": "sheet-id"})

    assert_success(result)
    assert_output_contains(result, "insights_text")


@pytest.mark.asyncio
async def test_missing_credential():
    from main import automation

    sandbox = LocalSandbox()
    sandbox.set_credentials(MockCredentialProxy([]))  # no services configured

    result = await sandbox.run(automation, {"spreadsheet_id": "sheet-id"})
    assert not result.success

Run tests:

torivers test               # run the full test suite
torivers test --coverage    # with coverage report

Cover at least: happy path, missing credential, malformed input, and one realistic edge case (empty dataset, rate-limit, oversized input). For LLM mocking, storage mocking, HTTP mocking, and all assertion helpers see references/testing-guide.md.

---

Phase 6 — Validate & Submit

6.1 — Pre-submit checklist

Before running torivers submit, verify every item:

• torivers validate --strict passes with zero errors and zero warnings
• torivers test passes; the suite covers happy path + at least one failure path
• README.md describes inputs, outputs, and a one-paragraph use case (not the template's placeholder text)
• DESIGN.md is up to date if the design shifted during implementation
• Pricing is sane: roughly 0.5 + 0.5 × num_llm_calls + 0.5 × num_credential_calls as a starting heuristic — adjust for value delivered, then sanity-check against similar marketplace listings
• No print() debug statements left in source files
• No hardcoded sample IDs or test data left in main.py or state.py
• Trigger-specific config matches the design: schedule_config for scheduled automations, and a webhooks: block for any automation that accepts incoming HTTP triggers

6.2 — Submit commands

torivers validate                           # already passing — confirm again
torivers validate --strict                  # stricter — runs all submission checks
torivers run --input-json '{"key":"val"}'   # local smoke test against real input shape
torivers submit --dry-run                   # preview without submitting
torivers submit -m "Initial submission"     # submit for marketplace review
torivers status                             # check review status

---

Sub-agent Collaboration

For non-trivial automations (3+ nodes, custom credentials, or substantial test coverage), parallelize the implementation phase by spawning sub-agents. A concrete split that works well:

• Agent 1 — implements state.py and node functions
• Agent 2 — implements tests/test_<name>.py against the agreed design block
• Agent 3 — drafts the marketplace README.md and reviews automation.yaml

Each agent gets the same DESIGN.md as its source of truth. You (the main session) integrate the results, run torivers validate, and resolve conflicts. Run them in the same message — they have no dependencies on each other once the design is locked.

For trivial automations (a single LLM call, no credentials), don't bother — the overhead of briefing three agents outweighs the parallelism.

---

Do's and Don'ts (Summary)

The full rationale for each rule, with explanations of why each one matters, is in references/anti-patterns.md. The condensed list:

Do:

• Import everything from torivers_sdk.* — no exceptions
• Log ctx.log_progress() before a step, ctx.log_success() or ctx.log_error() after
• Use pandas/numpy for all calculations — never delegate arithmetic to an LLM
• Declare every credential in the manifest (required_credentials / optional_credentials)
• Return only the changed keys from each node function
• Export automation = MyAutomation() at module level in main.py
• Use StorageClient for file I/O; HttpClient for outbound HTTP

Don't:

• Import any module not on the allowlist (torivers_sdk.*, the approved stdlib subset, or the approved third-party packages — see the capabilities snapshot)
• Use open() for file I/O or third-party HTTP libraries directly — use the SDK's StorageClient and HttpClient instead
• Hardcode secrets, API keys, or tokens in source files
• Call workflow.compile() inside build_graph()
• Feed raw data rows to an LLM — always summarize or aggregate first
• Access a credential service not listed in the manifest
• Apply retry policies to LLM-bound nodes — retries double-bill the user

---

Reference Files

File	Read when you need...
references/archetypes.md	The five common automation shapes — pick one in Phase 2
references/anti-patterns.md	The why behind each Do/Don't rule
references/capabilities-snapshot.json	Current credentials, models, categories, limits (offline fallback)
references/manifest-fields.md	All manifest fields, constraints, webhook/schedule configs
references/output-blocks.md	TextBlock, FileBlock, JsonDataBlock — examples and usage
references/patterns.md	Parallel fan-out, conditional routing, retry policies, async nodes
references/sdk-api.md	Method signatures and shapes for HttpClient, StorageClient, LLMClient, GmailClient, GoogleSheetsClient
references/testing-guide.md	Mock clients, LocalSandbox config, all assertion helpers
references/modifying-existing.md	Iterating on an existing project — read DESIGN.md / DECISIONS.md first

Scripts:

Script	Purpose
scripts/check-setup.sh	Verify torivers CLI is installed and the Python env is ready
scripts/fetch-capabilities.sh	Get live capabilities JSON; falls back to bundled snapshot