Sync Client
AgentTrustClient synchronous HTTP client for validate and audit operations.
Overview
AgentTrustClient is the synchronous HTTP client for calling the AgentTrust gateway. Use it when you need full control over validation requests, access to complete response objects, or integration with non-decorator code paths.
Why It Matters
The direct client exposes the full validation pipeline response — confidence scores, risk tiers, decision reasons, and envelope IDs — for dashboards, logging pipelines, and conditional routing.
Prerequisites
pip install agentrust-pyStep-by-Step Guide
1. Basic validation
from agentrust_sdk import AgentTrustClient
with AgentTrustClient() as client:
result = client.validate(
agent_id="my-agent",
user="alice",
input="User query text",
output={"answer": "Agent response"},
framework="Custom",
)
print(result.decision.outcome)
print(result.envelope_id)2. Constructor options
client = AgentTrustClient(
base_url="http://localhost:8000", # or AGENTRUST_GATEWAY_URL
api_key="at_team_...", # or AGENTRUST_KEY
timeout=10.0, # or AGENTRUST_TIMEOUT_SEC
raise_on_tier_gate=False, # raise TierGateError instead of warning
)Failure mode is env-only
There is no failure_mode constructor argument. The failure mode is read from
AGENTRUST_FAILURE_MODE on every call, so set it via the environment:
export AGENTRUST_FAILURE_MODE=closed. See Failure Modes.
3. Response fields
validate() returns a ValidateResponse. All scores are on a 0–100 scale.
result.envelope_id # audit record ID
result.validation.schema_score # 0–100
result.validation.tool_trust_score # 0–100
result.validation.policy_score # 0–100
result.validation.consistency_score # 0–100
result.validation.evidence_score # 0–100 or None when the signal is absent
result.validation.judge_score # 0–100 or None until the LLM judge runs
result.validation.final_confidence # 0–100 weighted confidence (Developer+)
result.validation.failures # list[str] — human-readable check failures
result.risk.tier # low | medium | high | critical (Developer+)
result.risk.score # 0–100 (Developer+)
result.risk.reason # str
result.decision.outcome # approve | retry | request_evidence | escalate | block | pending
result.decision.reason # str — why this outcome was chosen
result.decision.policy_version # str
result.latency_ms # gateway processing time
result.tier_info # "unknown" after a real gateway call; only the
# SDK's local paths set it ("disabled", "oss")
result.upgrade_hint # str | None
result.trust_chain # dict | None (Enterprise)Convenience properties:
result.approved # decision.outcome == "approve"
result.blocked # decision.outcome == "block"
result.needs_review # outcome in ("escalate", "request_evidence")
result.schema_valid # schema_score >= 80 and no failures — available on every tier4. OSS mode (no API key)
Without an API key, the client performs in-process schema validation only — no HTTP call.
decision.outcome stays "pending" and risk.tier stays "unknown", because those
engines are tier-gated. Check result.schema_valid instead.
5. Tool call metadata
tools_called takes a list of plain dicts (the client builds ToolCall models for
you). Field names are name, arguments, result, latency_ms, error.
result = client.validate(
agent_id="tool-agent",
user="bob",
input="Search database",
output={"rows": 5},
tools_called=[
{
"name": "sql_query",
"arguments": {"query": "SELECT ..."},
"result": {"rows": 5},
"latency_ms": 12.4,
},
],
)Entries that carry a result or an error are automatically forwarded to the gateway as
tool_results, which is what the tool-trust check scores against.
6. Two-phase gate: pre_check() / post_check()
validate() scores an execution that already happened. To stop a call before it runs —
what @harness does internally — use the pair:
with AgentTrustClient() as client:
pre = client.pre_check(
agent_id="payment-agent",
action="transfer_funds", # matched against the declared-tools manifest
params={"amount": 500, "recipient": "Bob"},
user="alice",
input="Transfer $500 to Bob",
)
if pre.blocked:
raise RuntimeError(pre.decision.reason) # the transfer never happens
result = do_transfer(amount=500, recipient="Bob")
post = client.post_check(
pre_check_envelope_id=pre.envelope_id,
output=result,
latency_ms=12.0,
)
if post.blocked:
raise RuntimeError(post.decision.reason) # governs release, not executionBoth return the same ValidateResponse shape. post_check() is idempotent — calling it
twice for one pre-check returns the first result instead of writing a duplicate audit row.
Examples
Conditional routing:
with AgentTrustClient() as client:
result = client.validate(agent_id="router", user=u, input=i, output=o)
if result.decision.outcome == "escalate":
send_to_human_review(result.envelope_id)
elif result.decision.outcome == "block":
return error_response(result.decision.reason)
return success_response(o)Best Practices
- Use context manager (
with AgentTrustClient()) for connection cleanup - Log
envelope_idwith every validation for audit correlation - Set
AGENTRUST_FAILURE_MODE=closedfor compliance-critical paths - Include
tools_calledwhen agents use tools (enables the tool-trust check)
Common Mistakes
- Creating a new client per request (use context manager or singleton)
- Ignoring
AGENTRUST_FAILURE_MODEwhen the gateway is down - Passing string output instead of dict (schema validation fails)
Troubleshooting
| Error | Cause |
|---|---|
GatewayUnavailableError | Gateway down + AGENTRUST_FAILURE_MODE=closed |
BlockedError | Decision outcome was block (raised by @harness, not by validate()) |
GatewayVersionError | SDK/gateway version mismatch |