Writing a verb

A verb is a typed Pydantic function decorated with @verb that becomes, at import time, three things at once:

1. A FastAPI route under /api/scidex/<name> (with . → /).

2. An MCP tool with the same input / output schemas.

3. A directly callable async function for in-process callers.

This is the v2 “one source, three outputs” promise from SPEC-001 §3.1. The registry in src/scidex_substrate/skill/framework.py is the single source of truth — there is no parallel HTTP-only or MCP-only path.

This tutorial walks through building a realistic new verb: scidex.paper.cite_count, which returns the number of artifacts that have an evidence_for or evidence_against link pointing at a given paper. It uses the existing paper artifact type and the existing artifact_links table — no new schema or migration.

1. Pick a name and decide read vs. write

Verb names are dotted (scidex.<area>.<operation>). The HTTP path is derived automatically by replacing dots with slashes: scidex.paper.cite_count → POST /api/scidex/paper/cite_count. The mapping lives in VerbSpec.http_path in framework.py.

Decide which @verb flags apply:

Flag	Meaning
read=True	The verb performs no DB writes and emits no events. The dispatcher does not call conn.commit() at the end and rate limits apply to the “read” bucket.
read=False (default)	The verb writes. The dispatcher wraps the call in a transaction and commits on success / rolls back on exception. Also triggers the write-metering hook (SPEC-178).
public=True	The HTTP route is reachable without a Bearer token, behind a 60 req/min per-IP sliding window. Only set this on safe read verbs (scidex.get, list, search, schema, comments, links).
batchable=True	The verb may appear inside a scidex.batch envelope (SPEC-062). Opt-in per verb.
requires_lock=True	For mutable_with_history and superseded_only types: the framework auto-injects a base_content_hash precondition check and returns a structured 409 with a JSONPatch delta on mismatch (SPEC-061).
signed=True	The HTTP path requires a verified X-Signed-Envelope header (SPEC-187). The dispatcher verifies, stamps WriteEnvelope.verified_signer_did, and validators trust it.

For cite_count we want read=True, public=True — it’s a cheap aggregation that anyone with the paper id should be able to call. Not batchable: it’s already cheap and the batching surface is for ops that benefit from server-side fan-out.

2. Write the file

Create src/scidex_substrate/skill/verbs/paper_cite_count.py.

"""scidex.paper.cite_count — count incoming evidence links for a paper.

A paper's evidence links are the polymorphic backbone of citation
analytics in the substrate. Hypotheses (and other artifacts) carry
``evidence_for`` / ``evidence_against`` links pointing at the papers
they rest on; counting those edges is the v2 equivalent of v1's
``paper.citation_count`` column without depending on a denormalized
field that could drift.

The verb is read-only, public, and cheap (one index-backed COUNT).
Per SPEC-001 §7.1 read verbs return a fixed Pydantic output model —
no envelope unless the caller asks for one.
"""

from __future__ import annotations

from typing import Literal

from psycopg.rows import dict_row
from pydantic import BaseModel, Field

from scidex_substrate.core.errors import SubstrateException
from scidex_substrate.skill.framework import verb
from scidex_substrate.skill.types import Context, Ref


class PaperCiteCountIn(BaseModel):
    ref: Ref | str = Field(
        description="The paper artifact. Object form, or stable string "
        "'paper:<id>'. Refs to non-paper types raise validation_failed.",
    )
    include_negative: bool = Field(
        default=True,
        description="When True (default), evidence_against edges count "
        "the same as evidence_for. When False, count only evidence_for.",
    )


class PaperCiteCountOut(BaseModel):
    ref: Ref
    total: int = Field(description="Total inbound evidence links.")
    by_predicate: dict[str, int] = Field(
        default_factory=dict,
        description="Per-predicate breakdown, e.g. "
        "{'evidence_for': 4, 'evidence_against': 1}.",
    )


@verb(name="scidex.paper.cite_count", read=True, public=True)
async def paper_cite_count(
    args: PaperCiteCountIn, ctx: Context
) -> PaperCiteCountOut:
    """Return the count of inbound evidence_* links for one paper."""
    target = args.ref if isinstance(args.ref, Ref) else Ref.parse(args.ref)
    if target.type != "paper":
        raise SubstrateException(
            "validation_failed",
            f"scidex.paper.cite_count requires a paper ref; got {target.type!r}",
        )

    predicates: tuple[str, ...] = (
        ("evidence_for", "evidence_against")
        if args.include_negative
        else ("evidence_for",)
    )

    sql = (
        "SELECT predicate, COUNT(*) AS n "
        "FROM artifact_links "
        "WHERE to_type = 'paper' "
        "  AND to_id = %s "
        "  AND predicate = ANY(%s) "
        "GROUP BY predicate"
    )
    breakdown: dict[str, int] = {}
    async with ctx.db().cursor(row_factory=dict_row) as cur:
        await cur.execute(sql, (target.id, list(predicates)))
        async for row in cur:
            breakdown[str(row["predicate"])] = int(row["n"])

    return PaperCiteCountOut(
        ref=target,
        total=sum(breakdown.values()),
        by_predicate=breakdown,
    )

A few things this snippet illustrates that are non-obvious:

• Ref | str. Every read verb that takes a ref accepts both the dataclass-style object and the canonical string. The Ref.parse helper handles the string form. See Ref in skill/types.py.

• Type-checking the ref. A polymorphic verb name like scidex.list accepts any registered type. A specialized verb like scidex.paper.cite_count should fail loud on a non-paper ref — raise SubstrateException("validation_failed", ...), not a bare ValueError, so the dispatcher returns a clean 400 with the standard error envelope.

• ctx.db() is the connection, not the pool. The dispatcher already opened a connection, set db_context, and (for writes) will commit when your function returns cleanly. Don’t open another connection; use the cursor pattern shown above.

• No await conn.commit() in a read verb. Reads are effectively wrapped in BEGIN ... ROLLBACK (no commit). For writes, the dispatcher commits — calling conn.commit() from your verb body would double-commit and lose the write-metering hook.

3. Wire it into the side-effect import

The FastAPI app loads verbs through one line in api/app.py:

import scidex_substrate.skill.verbs  # noqa: F401

That goes through src/scidex_substrate/skill/verbs/__init__.py, which imports every verb module by name. New files must be added to that __init__.py to register:

# src/scidex_substrate/skill/verbs/__init__.py
from . import (
    actor_create,
    actor_get,
    ...
    paper_cite_count,  # <-- add this
    ...
)

The @verb decorator runs at import time and populates the registry in framework.py. There is no separate “manifest” file to update.

4. Verify it registered

Restart your dev server (uvicorn ... --reload will pick up the new file). Then:

curl -s http://127.0.0.1:8200/ | jq '.verbs' | grep paper.cite_count

"scidex.paper.cite_count",

And exercise the route:

curl -s http://127.0.0.1:8200/api/scidex/paper/cite_count \
  -X POST \
  -H 'content-type: application/json' \
  -d '{"ref": "paper:p-abc123"}' | jq

If the paper exists with no inbound links you get:

{
  "ref":          {"type": "paper", "id": "p-abc123", "version": null},
  "total":        0,
  "by_predicate": {}
}

If you POSTed a non-paper ref:

{
  "error":      {
    "code":    "validation_failed",
    "message": "scidex.paper.cite_count requires a paper ref; got 'hypothesis'"
  },
  "details":    null,
  "request_id": "…",
  "retryable":  false,
  "retry_after_ms": null
}

The standard error envelope mapping lives in _status_for_code in api/app.py.

The verb is also now exposed as an MCP tool. If you have scidex-substrate-mcp running, a tools/list JSON-RPC call returns scidex.paper.cite_count with the input schema derived from PaperCiteCountIn.model_json_schema(). No extra wiring required.

5. Authoring tests

Per SPEC-007, every new verb needs at minimum:

• A unit test for the Pydantic input/output models (no DB).

• A service-layer test for the verb body called in-process.

• An HTTP integration test for the route and the auth path.

The repo’s pre-push hook (scripts/install-hooks.sh) runs pytest tests/unit automatically before push; the service and HTTP suites are part of the self-hosted CI flow.

Unit test (tests/unit/skill/verbs/test_paper_cite_count.py)

import pytest
from pydantic import ValidationError

from scidex_substrate.skill.types import Ref
from scidex_substrate.skill.verbs.paper_cite_count import (
    PaperCiteCountIn,
    PaperCiteCountOut,
)


def test_in_accepts_string_ref() -> None:
    args = PaperCiteCountIn.model_validate({"ref": "paper:p-1"})
    assert args.include_negative is True  # default
    assert args.ref == "paper:p-1"


def test_in_accepts_object_ref() -> None:
    ref = Ref(type="paper", id="p-1")
    args = PaperCiteCountIn(ref=ref, include_negative=False)
    assert args.ref == ref
    assert args.include_negative is False


def test_out_serializes_deterministically() -> None:
    out = PaperCiteCountOut(
        ref=Ref(type="paper", id="p-1"),
        total=3,
        by_predicate={"evidence_for": 2, "evidence_against": 1},
    )
    dumped = out.model_dump(mode="json")
    assert dumped["total"] == 3
    assert dumped["by_predicate"]["evidence_for"] == 2


def test_in_rejects_missing_ref() -> None:
    with pytest.raises(ValidationError):
        PaperCiteCountIn.model_validate({})

Service-layer test (tests/service/skill/verbs/test_paper_cite_count.py)

This is the layer where you exercise the actual verb body inside a transaction-scoped test DB. The substrate’s test conftest provides test_db_conn (per-test connection inside BEGIN; ... ROLLBACK;) and a test_context helper that builds a viewer-permission Context.

import pytest

from scidex_substrate.skill.framework import invoke
from scidex_substrate.skill.types import db_context


@pytest.mark.asyncio
async def test_cite_count_zero_for_unlinked_paper(test_db_conn, test_ctx):
    async with test_db_conn.cursor() as cur:
        await cur.execute(
            "INSERT INTO papers (paper_id, title) VALUES ('p-test1', 't')"
        )
    token = db_context.set(test_db_conn)
    try:
        result = await invoke(
            "scidex.paper.cite_count",
            {"ref": "paper:p-test1"},
            test_ctx,
        )
    finally:
        db_context.reset(token)
    assert result["total"] == 0
    assert result["by_predicate"] == {}


@pytest.mark.asyncio
async def test_cite_count_counts_both_predicates(test_db_conn, test_ctx):
    async with test_db_conn.cursor() as cur:
        await cur.execute(
            "INSERT INTO papers (paper_id, title) VALUES ('p-test2', 't')"
        )
        await cur.executemany(
            "INSERT INTO artifact_links "
            "(from_type, from_id, predicate, to_type, to_id) "
            "VALUES (%s, %s, %s, %s, %s)",
            [
                ("hypothesis", "h-a", "evidence_for", "paper", "p-test2"),
                ("hypothesis", "h-b", "evidence_for", "paper", "p-test2"),
                ("hypothesis", "h-c", "evidence_against", "paper", "p-test2"),
            ],
        )

    token = db_context.set(test_db_conn)
    try:
        result = await invoke(
            "scidex.paper.cite_count",
            {"ref": "paper:p-test2"},
            test_ctx,
        )
    finally:
        db_context.reset(token)
    assert result["total"] == 3
    assert result["by_predicate"] == {
        "evidence_for": 2,
        "evidence_against": 1,
    }

HTTP integration test (tests/integration/api/test_paper_cite_count_http.py)

import pytest
from httpx import AsyncClient


@pytest.mark.asyncio
async def test_cite_count_returns_200_and_envelope(app_client: AsyncClient):
    resp = await app_client.post(
        "/api/scidex/paper/cite_count",
        json={"ref": "paper:p-nonexistent"},
    )
    assert resp.status_code == 200
    body = resp.json()
    assert body["ref"]["type"] == "paper"
    assert body["total"] == 0


@pytest.mark.asyncio
async def test_cite_count_rejects_non_paper_ref(app_client: AsyncClient):
    resp = await app_client.post(
        "/api/scidex/paper/cite_count",
        json={"ref": "hypothesis:h-1"},
    )
    assert resp.status_code == 400
    body = resp.json()
    assert body["error"]["code"] == "validation_failed"

6. Common patterns

Write verb shape

A write verb is the same skeleton with read=False. The dispatcher opens a transaction, calls your body, then commits — so write side effects should be SQL-side. Read SPEC-001 §13 for the canonical scidex.create / scidex.update shape using WriteEnvelope.

@verb(name="scidex.my.create", read=False)
async def my_create(args: MyCreateIn, ctx: Context) -> MyCreateOut:
    ...
    # No conn.commit() — the dispatcher commits on a clean return.
    return MyCreateOut(...)

If your write needs the base_content_hash lock-token discipline, add requires_lock=True and put a base_content_hash field on the input model. The framework’s check_lock helper in framework.py gives you the structured 409 with JSONPatch delta on mismatch.

Permission gating inside the body

The dispatcher does not enforce permission strings; that’s the verb’s job. Pattern:

if not ctx.actor.id or ctx.actor.id == "anonymous":
    raise SubstrateException("permission_denied", "auth required")
if "contributor" not in ctx.actor.permissions:
    raise SubstrateException("permission_denied", "needs contributor")

The standard set of substrate error codes (and their HTTP mappings) is in _status_for_code in api/app.py. Use these codes — don’t invent new ones unless the rest of the codebase already uses them.

Emitting events

Write verbs that downstream agents care about should append to the event stream so SSE subscribers see them. The minimal pattern (from verbs/comment.py):

from scidex_substrate.skill import events as _events

await _events.publish(
    ctx.db(),
    event_type="paper.cite_count.queried",   # only do this for writes
    source=f"scidex.paper.cite_count/{ctx.actor.id}",
    payload={"paper_id": target.id, "total": total},
)

For pure read verbs (like cite_count) you usually do not publish events — read traffic would drown SSE subscribers.

7. References

• framework.py — @verb decorator, registry, invoke, check_lock.

• skill/types.py — Ref, Context, ActorRef, ArtifactEnvelope, WriteEnvelope, WriteResult.

• api/app.py — attach_verb_routes, the dispatcher, _status_for_code, the string-annotation footgun documented in _make_handler.

• verbs/actor_create.py — write verb with permission gating + ON CONFLICT idempotency.

• verbs/comment.py — write verb with event publication + mention extraction.

• verbs/search.py — a complex read verb with mode dispatch and the public=True flag.

• SPEC-001 §7 — the verb-surface design.

• SPEC-007 — six-layer test strategy.

• SPEC-061 — when to set requires_lock=True.

• SPEC-062 — batchable=True.

• SPEC-187 §5 (signed envelope verification) — when to set signed=True. The spec is referenced extensively from framework.py and from WriteEnvelope.verified_signer_did; reach for any existing signed=True verb in skill/verbs/ for a worked example.