agentvalet

Call approved SaaS platforms from any Python agent. Your agent never holds the downstream credential — AgentValet signs a short-lived identity assertion, checks the call against the owner's grants and policy, injects the credential at call time, and writes an audit record.

This is the Python port of @agentvalet/client. Same endpoints, same approval semantics, same timings.

Install

pip install agentvalet

Python 3.9+. Two dependencies: httpx and pyjwt[crypto].

Get an agent identity

agentvalet register --code <invite-or-enrollment-code>

The RSA keypair is generated on your machine. Only the public half is sent; the private key is written to ~/.agentvalet/agent.key (mode 0600) and never crosses the wire. The code is either an invite's bind secret or the enrollment code from the "Try it free" flow.

This writes the same files as npx @agentvalet/register, so the Node and Python tooling are interchangeable — you do not need Node installed.

Use it

from agentvalet import AgentValet

av = AgentValet.from_env()

result = av.call(
    platform="slack",
    endpoint="/api/chat.postMessage",
    method="POST",
    scope="chat:write",
    data={"channel": "#general", "text": "Deploy finished."},
)

call() returns the broker's envelope — the upstream SaaS body under data, with _meta describing the call itself:

{
    "data": {"ok": True, "ts": "1723800000.000100"},   # exactly what Slack returned
    "_meta": {"capability": "agent.action", "timestamp": "…", "next_actions": [...]},
}

So it's result["data"]["ok"], not result["ok"]. The split is deliberate: data is the upstream payload byte-for-byte, so anything the broker adds about the call stays outside it.

from_env() reads AGENTVALET_AGENT_ID / AGENTVALET_OWNER_ID (or the bare AGENT_ID / OWNER_ID) and finds the key via AGENT_PRIVATE_KEY_B64, AGENT_PRIVATE_KEY_PATH, AGENT_PRIVATE_KEY, or ~/.agentvalet/agent.key. To wire it explicitly:

av = AgentValet(
    agent_id=os.environ["AGENT_ID"],
    owner_id=os.environ["OWNER_ID"],
    private_key=os.environ["AGENT_PRIVATE_KEY"],
    proxy_url="https://api.agentvalet.ai",   # default
)

Use it as a context manager to close the HTTP client cleanly:

with AgentValet.from_env() as av:
    av.call(platform="github", endpoint="/user", scope="read")

Async

AsyncAgentValet is a method-for-method mirror. All decision logic is shared, and the test suite runs the same behaviour table against both.

from agentvalet import AsyncAgentValet

async with AsyncAgentValet.from_env() as av:
    result = await av.call(
        platform="slack", endpoint="/api/chat.postMessage",
        method="POST", scope="chat:write", data={...},
    )

Approvals are just a slower call

When the owner has marked a scope as approval-gated, the proxy holds the action and call() waits. If the owner approves, the proxy re-runs the call and you get the result — from your code's point of view it simply took longer.

av = AgentValet.from_env(
    on_approval_pending=lambda i: print(f"waiting… {i['elapsed_s']:.0f}s")
)

If nobody responds inside the budget (50s by default), you get an ApprovalTimeoutError — not a failure. The action stays queued. Keep the approval_id and resume whenever you like, in this process or another:

from agentvalet import ApprovalTimeoutError

try:
    av.call(platform="stripe", endpoint="/v1/refunds", method="POST", scope="charge")
except ApprovalTimeoutError as err:
    queue.put(err.approval_id)          # hand off to a worker

# …later, elsewhere:
result = av.wait_for_approval(approval_id)

Pass approval_timeout_s=0 if you never want to block.

Errors you can branch on

Every exception is typed, so you never string-match an error envelope:

Asking for access you don't have

Deny-by-default means a scope you were never granted raises AccessDeniedError. Your agent can ask for it:

result = av.request_access(
    platform="slack",
    scope="chat:write",
    reason="Post deploy notifications to #general",
)
if result["status"] == "approved":
    ...  # retry the original call

Checking before you act

av.list_platforms()                 # what this agent is actually granted
av.pending_actions()                # queued behind an approval
av.evaluate("stripe", "charge")     # dry-run the decision, no side effect

evaluate() is worth calling before anything destructive.

Which connection served the call

When a platform has several connections and you don't pin one, the broker uses the default. That's the quiet failure behind a lot of confused agents: you ask for a repo, the default GitHub account doesn't have it, you get a 404, and you conclude the repo doesn't exist rather than that you're on the wrong account.

When there was more than one connection you could have used, the response says so. On success it rides _meta, leaving data untouched:

res["_meta"]["connection"]
# {
#   "used": "o-github", "label": "Acme", "defaulted": True,
#   "others": [{"connection_id": "o-github-9f1c", "label": "Personal"}],
#   "hint": "…the target may live on another connection — retry with connection_id…",
# }

On an upstream error there's no envelope — the body is exactly what the platform sent — so the same information arrives as one namespaced key, carried on ProxyError.body:

{
  "message": "Not Found",
  "_agentvalet": { "connection": { "used": "o-github", "defaulted": true, "others": [], "hint": "…" } }
}

The upstream keys stay where they were, so existing error handling is unaffected. It's absent when there's nothing to act on — a connection you pinned, a platform with one connection, or a lookup that failed — so treat its absence as "no alternatives", never as "this is definitely the right account".

Subagents (child identities)

A supervisor agent can mint a time-boxed, scope-attenuated identity for a worker it spawns — without ever handing over its own keys. Give the worker only bearer_token; it builds its own client with from_bearer():

from agentvalet import AgentValet

supervisor = AgentValet.from_env()

# Attenuated server-side against the supervisor's own grants — a worker can
# never end up with more access than its supervisor has.
child = supervisor.issue_child(
    name="worker-a",
    ttl_seconds=900,                        # 60..3600, default 900
    grants=[{"platform": "slack", "scopes": ["chat:write"]}],
)

# Hand the worker child.bearer_token, not the supervisor's identity or key.
worker = child.client()                     # == AgentValet.from_bearer(child.bearer_token)
worker.call(platform="slack", endpoint="/api/chat.postMessage",
            method="POST", scope="chat:write", data={...})

A bearer-mode client behaves like any other — call, list_platforms, evaluate, request_access all work unchanged — except it authenticates with the token verbatim instead of signing a fresh assertion, and it cannot itself call issue_child(): the server enforces delegation depth 1, so a child can never mint a grandchild. Suspending child.child_agent_id kills only that worker; sibling workers, each with their own agent id and token, keep running. See examples/05_langgraph_subagents.py for a full supervisor/worker pattern with LangGraph.

Self-hosting

Point proxy_url at your own deployment. Everything else is identical.

License

MIT