Python SDK
The complete veefore-py surface — guard, protect, the client, dataclasses and typed exceptions.
bashpip install veefore-py
The package is veefore-py; the module you import is veefore. Python 3.9+.
One dependency: cryptography, for Ed25519 signing.
Running veefore in your project installs this package with your project's
tool (pip, uv or poetry), writes its configuration and adds @guard to your
tools. See Automatic setup.
A convenience layer, never a security boundary
Every protected action is decided server-side by the gateway on every call. Nothing in this package can grant an agent authority it does not already have, and nothing here caches a permission.
The Python and TypeScript SDKs produce byte-identical canonical payloads and signatures.
#@guard(tool_key, **options)
The way most agents integrate. Decorate the function that performs a tool, and every call is decided by the gateway before the function runs.
pythonfrom veefore import guard
from langchain_core.tools import tool
@tool
@guard("refund_customer")
def refund_customer(order_id: str, amount: int) -> str:
"""Refund a customer order."""
return payments.refund(order_id, amount)
Put @guard directly above def, below the framework's decorator. It uses
functools.wraps, so @tool, @function_tool and @mcp.tool() still see the
original name, signature, type hints and docstring. async def functions work
too; the gateway call runs off the event loop.
| Decision | What @guard does |
|---|---|
Permitted, CLIENT tool |
Runs the function and returns its result |
Permitted, brokered HTTP tool |
Returns the gateway's result without running the function |
| Refused | Raises VeeforeBlockedError |
| Needs a human | Raises VeeforeApprovalRequiredError, or waits with wait_for_approval |
The call's arguments, by name, are sent as the policy parameters.
capabilitystroptionalDefaults to
tool.<tool_key>, whichveefore initregisters.wait_for_approvalbool or floatoptionalWait for a human instead of raising.
Truewaits up to 900 seconds; a number sets the timeout in seconds.describeCallable[..., dict]optionalReturns extra policy inputs for a call, such as
amount,currency,destinationorpurpose.clientVeeforeClientoptionalUse this client instead of the shared one built from the environment.
Configuration comes from the VEEFORE_* environment variables. If they are not
set, @guard reads the block veefore init writes to .env.local; values
already in the environment always win.
#protect(agent, **options)
For code that calls Veefore directly rather than through a tool function.
pythonfrom veefore import protect
agent = protect(my_agent, name="finance-agent")
Returns a ProtectedAgent wrapping your object. Attribute access falls through
to the original; veefore, act, evaluate, simulate and handshake are
added.
namestroptionalDisplay metadata. Never used for authorization.
gateway_urlstrrequiredFrom
VEEFORE_GATEWAY_URLif not passed.agent_idstrrequiredFrom
VEEFORE_AGENT_ID.private_keystrrequiredBase64 PKCS#8 Ed25519. From
VEEFORE_AGENT_PRIVATE_KEY.key_idstrrequiredFrom
VEEFORE_AGENT_KEY_ID.audiencestroptionalMust match the gateway's published audience exactly. Defaults to
gateway_url.declared_toolslist[str]optionalReported at handshake, so an unauthorized tool is a startup warning.
eager_handshakebooloptionalOpens the session immediately. Defaults to
False.request_timeout_secondsfloatoptionalPer-request timeout. Note: seconds, where the TypeScript SDK uses milliseconds.
max_retriesintoptionalTransport errors only. Decisions are never retried.
on_logCallableoptionalStructured diagnostics. Never receives secrets.
Missing required configuration raises VeeforeConfigurationError immediately.
#create_client() and config_from_env()
pythonfrom veefore import config_from_env, create_client
client = create_client(name="worker")
partial = config_from_env()
Use create_client when there is no agent object to wrap.
#VeeforeClient
pythoninfo = agent.handshake()
info.mode # 'OBSERVE' | 'SIMULATE' | 'ENFORCE'
info.capabilities
info.authorized_tools
info.passport_version
info.expires_at
agent.veefore.authenticate()
agent.veefore.info # last HandshakeResult, or None
| Method | Behaviour |
|---|---|
act(request) |
Performs the action. Raises when refused |
evaluate(request) |
Returns the decision as data |
simulate(request) |
evaluate with dry_run=True |
act_with_approval(request, *, timeout_seconds=900.0) |
Waits for a human, then resubmits |
approval_status(approval_id) |
Current status |
wait_for_approval(approval_id, ...) |
Polls, honouring poll_after_ms |
mcp_call(...) |
Invoke an authorized MCP tool |
a2a_invoke(...) |
Invoke a verified A2A peer |
#ActionRequest
A frozen dataclass.
pythonfrom veefore import ActionRequest
request = ActionRequest(
tool="refund_customer", # required
action="create",
capability="refund.create",
parameters={"orderId": "ord_8812", "reason": "damaged on arrival"},
resource={"key": "ord_8812", "type": "order"},
amount=250_00, # minor units; a policy input only
currency="INR",
destination="payments.internal.example.com",
purpose="Customer reported damage; agent verified photo evidence.",
idempotency_key="refund-ord_8812-attempt-1",
approval_id=None,
dry_run=None,
metadata=None,
)
Fields use snake_case; the SDK converts to the wire format. Fill in purpose
— it is what an approver reads.
#ActionDecision
pythondecision = agent.act(request)
decision.allowed # bool — branch on this
decision.decision # 'ALLOW' | 'BLOCK' | 'REQUIRE_APPROVAL' | …
decision.outcome
decision.mode
decision.result # Optional[dict]
decision.reason
decision.explanation # dict
decision.approval # Optional[dict]
decision.risk # Optional[dict]
decision.shadow # Optional[dict]
decision.limits # Optional[dict]
decision.latency_ms
decision.stages # list[StageResult]
decision.action_id
decision.request_id
decision.trace_id
Nested fields are plain dicts
approval, risk, shadow, limits and explanation are dict, not
dataclasses, because their shapes are extended server-side over time and a
frozen dataclass would reject a new field. Use decision.approval["approval_id"]
rather than attribute access.
stages is the exception — a list of StageResult dataclasses, since that shape
is fixed.
#Exceptions
| Exception | Meaning | Retry |
|---|---|---|
VeeforeConfigurationError |
Required configuration missing | no |
VeeforeAuthenticationError |
Identity refused | no |
VeeforeBlockedError |
Policy refused. Carries .decision |
never |
VeeforeApprovalRequiredError |
Human needed. Carries .approval_id |
wait, do not retry |
VeeforeApprovalDeniedError |
A human said no. Carries .approval_id |
no |
VeeforeRateLimitError |
429 |
yes, with backoff |
VeeforeConnectionError |
Transport failure | yes, with backoff |
VeeforeError |
Base class | depends |
pythonfrom veefore import VeeforeApprovalRequiredError, VeeforeBlockedError
try:
decision = agent.act(request)
except VeeforeApprovalRequiredError as error:
queue_for_human(error.approval_id)
except VeeforeBlockedError as error:
report_refusal(str(error), error.decision)
#Assertion helpers
pythonfrom veefore import (
ASSERTION_TTL_SECONDS,
canonical_json,
create_assertion,
generate_nonce,
)
canonical_json matches the server's canonicalisation exactly — sorted keys, no
insignificant whitespace — which is what makes cross-language signing work.
#Synchronous, deliberately
The client is synchronous. A protected action is a single round trip whose latency is dominated by the server, and the SDK is most often used inside a framework that already manages its own concurrency.
For async code, run it in a thread:
pythonimport asyncio
decision = await asyncio.to_thread(agent.act, request)
That keeps one well-tested code path rather than two that can drift.
#Differences from the TypeScript SDK
| TypeScript | Python | |
|---|---|---|
| Request fields | camelCase |
snake_case |
| Timeouts | requestTimeoutMs, timeoutMs |
request_timeout_seconds, timeout_seconds |
| Nested decision fields | typed interfaces | plain dict |
| Concurrency | async | synchronous |
The wire format, the canonicalisation, the signatures and the server behaviour are identical. There is no capability one SDK has and the other does not.