Skip to content
Vectadyne

Python

The distribution is vectadyne-agent; the import is vectadyne.

Terminal window
pip install vectadyne-agent

Same seven beats as the Go and TypeScript tutorials, in the same order.

1. Point at the endpoint

Section titled “1. Point at the endpoint”
import os
from vectadyne import Vectadyne


with Vectadyne(
    mcp_url=os.environ["VECTADYNE_MCP_URL"],
    token=os.environ["VECTADYNE_TOKEN"],
) as vd:
    ...

Two imports are not re-exported from the top level, and both are easy to trip over because the obvious guess fails:

from vectadyne.auth import Secret          # NOT vectadyne.Secret
from vectadyne.aio import AsyncVectadyne   # NOT vectadyne.AsyncVectadyne

Secret wraps a credential so that formatting it — repr, asdict, vars, an f-string — yields <redacted>. Reading the real value is an explicit .reveal(), which is greppable in review. Redaction is a property of the value, not of one display path.

AsyncVectadyne is the async facade. It mirrors the sync one method for method, with the same parameter names and the same typed return values — a parity the test suite pins, because an earlier version returned dicts from one and DTOs from the other and every async example in the README raised AttributeError.

2. Remember and recall

Section titled “2. Remember and recall”
ack = vd.remember(
    "The nightly reconcile job fails when billing/reconcile.py is edited without regenerating fixtures.",
    kind="lesson",
    scope_key="repo:payments",
)


hits = vd.recall("reconcile job failures", top_k=8)
for h in hits:
    print(h.content, h.memory_id)      # typed — attributes, not dict keys

Exactly one of content or messages. The SDK checks it locally so a typo is an error at the call site rather than a round trip.

An empty recall is a legitimate answer: results are admission-filtered at read time, so “nothing you may see” and “nothing exists” are indistinguishable on purpose.

3. Assemble

Section titled “3. Assemble”
assembled = vd.assemble(
    action_type="code.apply_patch",
    summary="Rewrite the reconcile loop",
    changed_files=["billing/reconcile.py"],
    scope_key="repo:payments",
    purpose="refactor",
)

Guides, memory, verdict and citations in one round trip, consistent with each other.

4. Branch on the verdict

Section titled “4. Branch on the verdict”

The polarity is inverted between our SDKs. Python is the odd one out. Do not port a snippet from another language by eye:

SDKFieldtrue means
GoGuidance.Holdstop
TypeScriptguidance.holdstop
PythonGovernanceDecision.proceedgo
from vectadyne import interpret_verdict


d = interpret_verdict(assembled.verdict)
if not d.proceed:
    print(d.summary())
    return


for w in d.warnings:
    print("advisory:", w)     # surface to the model; do not obey
  • A warn proceeds. warn is advice, not a stop. Treating it as a stop rebuilds the inline gate Vectadyne is designed not to be.
  • An unknown decision sets unknown and does not proceed. All three SDKs fail closed on a vocabulary they cannot interpret.

5. Split errors four ways

Section titled “5. Split errors four ways”
from vectadyne import (
    AuthError, ValidationError, RateLimitedError, TransportError, ToolError,
)


try:
    assembled = vd.assemble(...)
except AuthError:
    ...        # refresh the credential; never retry in a loop
except ValidationError:
    ...        # the request is wrong; fix the call site
except RateLimitedError as e:
    ...        # e.retry_after is the server telling you when the bucket refills
except TransportError:
    ...        # the platform did NOT answer — this is not a refusal

RateLimitedError carries retry_after typed rather than buried in a message string, because it is the one field that says what to do next.

The distinction to hold on to: “the platform said no” is not “the platform did not answer.” The SDK never invents a verdict, so an outage is a typed transport error and proceeding anyway is a line you write explicitly. See Handle errors.

6. Close the loop

Section titled “6. Close the loop”
vd.client.call_tool("evidence.ingest", evidence_args(
    agent="release-bot",
    items=[{
        "source_type": "pr",
        "summary": "Reconcile loop rewritten; fixtures regenerated",
        "scope_key": "repo:payments",
    }],
))

And measure what the context cost:

from vectadyne import ContextMeter


meter = ContextMeter(agent="release-bot", window_total=128_000)
meter.observe_assemble(assembled)
meter.set("system_prompt", 1_200)
meter.submit(vd.client)

An unreported segment is stored absent, never zero.

Production: a machine credential

Section titled “Production: a machine credential”

Python is where most CI integrations start, so this beat is here rather than in a guide. Do not put a person’s bearer token in a pipeline secret:

from vectadyne import AgentConfig, ClientCredentials
from vectadyne.auth import Secret


cfg = AgentConfig(
    mcp_url=os.environ["VECTADYNE_MCP_URL"],
    token_source=ClientCredentials(
        client_id=os.environ["VECTADYNE_CLIENT_ID"],
        client_secret=Secret(os.environ["VECTADYNE_CLIENT_SECRET"]),
    ),
)

The SDK exchanges the credential for a token and refreshes it transparently. A 429 from the token endpoint uses the OAuth error shape, not our fault envelope — the SDK normalises both into the same typed RateLimitedError so you do not have to parse two. See Enrol a machine principal.

7. Test that you asked

Section titled “7. Test that you asked”
from vectadyne.testing import FakeControlPlane


with FakeControlPlane() as fake:
    fake.verdict_for("code.apply_patch", "block", "fragile file")
    # … run your agent against fake.mcp_url …
    fake.assert_called("preflight.action")
    assert fake.request_count == 1     # a refusal is not retried

Assert the request count. A block must produce exactly one request.

Reference

Section titled “Reference”