Python

agentvalet is the Python port of @agentvalet/client. Same endpoints, same approval semantics, same timings. A LangChain tool, a Celery task, a FastAPI service, a script that reconciles invoices at 3am: anything that isn’t an MCP host gets the same governance, without hand-rolling RS256 JWTs.

If you’re writing a CrewAI crew, start at CrewAI instead — crewai-agentvalet builds the tools for you and uses this package underneath.

Install

pip install agentvalet

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

Then register an agent, if you haven’t already:

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 at mode 0600 and never crosses the wire. The code is either an invite’s bind secret or the enrollment code from the free trial flow.

This writes the same files as npx @agentvalet/register, so the Node and Python tooling are interchangeable — you don’t need Node installed to use Python.

Your first call

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

There is no Slack token in that file, in your environment, or in the model’s context. The SDK signs a 60-second identity assertion, the broker checks the call against your grants and policy, decrypts the real credential in memory, makes the call, and writes an audit row.

call() returns the broker’s envelope, parsed — 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": [...]},
}

The split is deliberate: data is the upstream payload byte-for-byte, so anything the broker wants to tell you about the call stays outside it.

The client is a context manager because it holds an httpx.Client; use with, or call close() yourself when you’re done.

Call arguments

Where it finds your identity

from_env() works with no arguments on a machine that has been through the register flow. Both naming conventions are accepted, and the AGENTVALET_-prefixed name wins when both are set.

The private key is resolved in this precedence order:

1. AGENT_PRIVATE_KEY_B64 — base64-encoded PEM. This is the container-safe one
2. AGENT_PRIVATE_KEY_PATH — path to a PEM file
3. AGENT_PRIVATE_KEY — raw multi-line PEM, or a single line with escaped newlines
4. ~/.agentvalet/agent.key — written by the register flow

Configuring the client explicitly

from_env() takes keyword overrides, or construct it directly:

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",
)

Note the unit. The TypeScript client uses milliseconds; Python uses seconds, matching the httpx convention. The underlying values are identical.

The 50-second approval budget sits deliberately under Claude Desktop’s 60-second tool timeout, so the SDK path and the MCP path give up at the same moment.

Which connection served the call

When a platform has more than one connection and you don’t pin one, the broker picks the default. That’s the quiet failure mode 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.

So when there was more than one connection you could have used, the response says so.

On success it rides _meta, leaving data untouched:

res = av.call(platform="github", endpoint="/repos/acme/x", scope="repo:read")
 
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 comes back exactly as the platform sent it — so the same information arrives as one namespaced key, and ProxyError.body carries it:

{
  "message": "Not Found",
  "documentation_url": "https://docs.github.com/...",
  "_agentvalet": { "connection": { "used": "o-github", "defaulted": true, "others": [...], "hint": "..." } }
}

The upstream keys stay exactly where they were, so your existing error handling is unaffected.

It’s absent whenever there’s nothing to act on: a connection you pinned yourself, a platform with only one connection, or a lookup that failed. Treat its absence as “no alternatives”, never as “this is definitely the right account”.

Approvals are a slower call

When your policy marks a scope as needing a human, nothing about your code changes. The broker returns 202 with an approval id, the SDK enters a long-poll, and you approve from the dashboard, an email or your phone. The call then executes and call() returns its result. From your program’s point of view it simply took longer.

av = AgentValet.from_env(
    on_approval_pending=lambda info: print(
        f"waiting {info['elapsed_s']:.0f}s on {info['platform']}:{info['scope']}"
    ),
)

The callback receives a dict with approval_id, elapsed_s, platform, scope and endpoint. On AsyncAgentValet it may be a coroutine function or a plain one.

When nobody answers

Past the budget you get ApprovalTimeoutError, and this is not a failure. The action is still queued server-side and will run if the owner approves. Hold the approval_id and resume later, from this process or a completely different one:

from agentvalet import ApprovalTimeoutError
 
try:
    av.call(platform="slack", endpoint="/api/chat.postMessage",
            method="POST", scope="chat:write", data=message)
 
except ApprovalTimeoutError as err:
    save_for_later(err.approval_id)      # resume in another run
 
# later, anywhere:
result = av.wait_for_approval(approval_id)

Resume rather than retry. Calling call() again queues a second copy of the action.

Setting approval_timeout_s=0 turns every approval-gated call into an immediate ApprovalTimeoutError carrying the id, which is what you want in a worker that shouldn’t sit blocked for a minute.

Errors you can branch on

Governance you can’t handle in code is just an outage. Every exception the package raises is one of these, so your agent can tell “I’m not allowed to” apart from “Slack is down” apart from “a human said no”, without string-matching an error envelope.

All of them subclass AgentValetError, so one except can separate broker problems from everything else in your process.

Asking for access you don’t have

AccessDeniedError is recoverable. request_access() asks an org admin to grant the scope and polls for the decision:

from agentvalet import AccessDeniedError
 
try:
    av.call(platform="stripe", endpoint="/v1/refunds", method="POST",
            scope="refunds:write", data={"charge": charge_id})
 
except AccessDeniedError as err:
    decision = av.request_access(
        platform=err.platform,
        scope=err.scope,
        reason="Refund duplicate charges flagged by support",
    )
    # decision["status"] is "approved" | "denied" | "pending"

On "approved", retry the original call(). On "pending" the request is still open — the admin hasn’t decided yet.

See Access requests for what the admin sees.

Finding out what you’re allowed to do

Three read-only calls, none of which perform an action:

grants  = av.list_platforms()                      # platforms + scopes actually granted
pending = av.pending_actions()                     # queued behind an approval
check   = av.evaluate("stripe", "refunds:write")   # dry-run a decision

evaluate() is worth calling before anything destructive. It tells you whether the action would be allowed, with no side effect and no audit consequence for having asked.

list_platforms() is the deny-by-default surface: if a platform and scope aren’t in there, call() will raise AccessDeniedError.

Async

AsyncAgentValet mirrors the sync client method for method, on httpx.AsyncClient. Use it in FastAPI, or anywhere blocking for 50 seconds on an approval would tie up an event loop.

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={"channel": "#general", "text": "Deploy finished."},
    )

call, wait_for_approval, list_platforms, pending_actions, evaluate and request_access are all awaitable. sign_jwt() stays synchronous — it’s pure CPU. Close it with aclose() if you’re not using async with.

Notes for production

Assertions live 60 seconds. The SDK signs a fresh one per request, so this is invisible to you — but it’s why any integration that pins a bearer token at construction time breaks a minute into a run. Notably, CrewAI’s MCPServerHTTP does exactly that; see CrewAI for why the stdio server is the right choice there.

The key is the only secret on the box. Mount it as a file and point AGENT_PRIVATE_KEY_PATH at it, or pass AGENT_PRIVATE_KEY_B64 in a 12-factor environment. Nothing else needs to travel with your agent, and no platform credential ever does.

Repeated auth failures trip the breaker. A wrong key doesn’t fail quietly forever — see Circuit breaker for what suspends an agent and how to clear it.

Next steps

• Scopes and permissions — how grants are shaped, and why the scope string must match verbatim
• Approval flow — what an approver actually sees
• Audit and revocation — killing an agent that misbehaves
• CrewAI — governed tools built from your grant matrix
• Node client — the same API, in TypeScript