Forking this into your product

There is no domain code to strip — the envelope and the buckets are already neutral. The work is wiring the two primitives into your tool surface.

1. Stamp at every tool return boundary

Wrap whatever your handler returns. The wrap is non-destructive, so existing fields pass through.

from mcp_tool_provenance import stamp, make_tags, ToulminArgument, QualifierLevel, DataAvailability

def read_records(args):
    rows = _do_read(args)            # your real work
    return stamp(
        {"records": rows, "count": len(rows)},
        tool_name="read-records",
        toulmin=ToulminArgument(
            claim=f"{len(rows)} records matched",
            grounds="The query key resolved against the primary store",
            warrant="A key matching N rows returns those N rows",
            qualifier=QualifierLevel.CERTAIN,
            rebuttal="A row could be soft-deleted after this read",
            data_availability=DataAvailability.SUFFICIENT,
        ),
        tags=make_tags(scope="tenant_specific", signal_type="operational",
                       attribution="read-records-tool", uncertainty="low"),
    )

If you prefer not to build the value objects, pass plain dicts — toulmin= accepts any five-key mapping and tags= any four-key mapping.

Stamp once, centrally

Rather than calling stamp in every handler, put it in your dispatch chokepoint:

def dispatch(tool_name, args):
    result, toulmin, tags = REGISTRY[tool_name](args)   # handlers return the triple
    return stamp(result, tool_name=tool_name, toulmin=toulmin, tags=tags)

Now no tool can return an un-stamped result. Add assert is_stamped(out) in tests for belt-and-suspenders.

2. Classify your surface and gate it

Maintain one classification next to your registry, then assert it in CI.

# your_app/surface.py
from mcp_tool_provenance import triage

REGISTRY = (...)   # the source of truth — e.g. tuple(ADAPTERS.keys())

SURFACE = triage(
    real_dispatch=("create-record", "update-record", "delete-record", "send-message"),
    read_only=("read-record", "list-records", "search-records"),
    native_only=("draft-guidance", "explain-plan"),
)

# tests/test_surface_drift.py
from mcp_tool_provenance import assert_no_drift
from your_app.surface import REGISTRY, SURFACE

def test_surface_is_mece():
    assert_no_drift(REGISTRY, SURFACE)   # fails if any tool is unclassified or in two buckets

The day someone adds a tool to REGISTRY without classifying it, this test goes red — which is exactly the anti-inflation property you want.

3. Publish the honest count

Wherever you advertise the surface (a list_tools-style endpoint, a README), report the derived split, not a single number:

from mcp_tool_provenance import surface_summary
summary = surface_summary(REGISTRY, SURFACE)
# summary["executes_server_side"]  -> tools that really act
# summary["native_only_count"]     -> guidance-only tools
# summary["unclassified"]          -> must be []  (built-in drift guard)

What NOT to change

The three bucket names (real_dispatch, read_only, native_only) — they are the MECE contract drift_check enforces.
The envelope key names (_toulmin, _tags, and the five/four sub-keys) — they are what makes results queryable uniformly across the surface.
The "stamp returns a new dict" discipline — never reach in and mutate a payload; rely on the non-destructive wrap so concurrent callers can't see a half-stamped result.

FilesExpand file tree

FORKING.md

Latest commit

History