agenttrafficlab

agenttrafficlab

Zero-dependency Python client for Agent Traffic Lab, an execution infrastructure layer for supported external agent tasks. ATL handles eligible provider selection, execution, bounded fallback, result verification, and durable Outcome when supply is available.

Install

pip install agenttrafficlab

Recommended usage: one call → result → automatic Outcome

from agenttrafficlab import run

result = run("search latest critical CVE")
print(result["result"])
print(result["execution_status"])

For the default tenant, run() now calls ATL's canonical atl_complete_task MCP tool directly. The caller sends the natural-language task once; ATL chooses an eligible provider/tool, executes it, applies bounded fallback when needed, verifies the result, and records the durable Outcome automatically. There is no client-side decide -> execute round trip and no separate feedback callback.

When capability is omitted, the SDK leaves capability inference to ATL. Pass it only when the caller already knows the required capability.

result = run("extract the fields from this page")
result = run("custom task", capability="extract")

You can also call the default path explicitly:

from agenttrafficlab import complete_task

result = complete_task("translate Hello into Spanish")

The returned object is ATL's structured task result and retains advanced decision and execution evidence for callers that need it.

Optional: recover a failed primary call

If an external provider fails, an agent host can explicitly opt into a bounded rescue. This is for user-authorized read-only or idempotent work only, not purchases, writes or other non-repeat-safe actions. A failure does not mean the original tool had no side effects.

from agenttrafficlab import run_with_rescue

def primary():
    return existing_provider_summary()  # your own existing function

result = run_with_rescue(
    "Summarize this supplied support ticket: Unable to sign in after a reset.",
    primary,
    capability="summarize",
    recoverable=(TimeoutError, ConnectionError),
)
print(result["source"], result["result"])

If the primary succeeds, no ATL request occurs. Only the specified exception classes trigger rescue; unrelated failures propagate. An optional accept_primary=lambda result: ... can reject an invalid primary output and trigger rescue. ATL receives only the task supplied by the caller, not the exception or stack. Rescue occurs at most once and succeeds only when ok=true, execution_status=EXECUTED, and result_validation.passed=true; otherwise the helper raises ATLClientError. Host permissions, provider eligibility and billing boundaries still apply.

For normal tasks without any failure, continue using complete_task() or run() directly when ATL's routing, validation and Outcome add value. The host decides whether to delegate; ATL cannot force calls from unrelated agents.

Advanced: Decision only

from agenttrafficlab import decide

result = decide("search latest critical CVE")
print(result["recommended_provider"]["provider_id"])

Use decide() only when your own system specifically needs routing/Decision evidence without immediate ATL-owned completion. execute_decision() remains available for explicit Decision-bound execution. If you pass a non-default tenant_scope to run(), the client preserves that advanced Decision-bound path for backward compatibility.

When execution happens outside ATL, report the real result if you want it included as Outcome evidence. Normal run() / complete_task() usage records Outcome automatically.

Provider selection remains ATL-owned; callers do not choose a provider directly.

Public surfaces:

• MCP / default complete-task path: https://mcp.agenttrafficlab.com/mcp
• Decision API: https://mcp.agenttrafficlab.com/intent-entrance/decision
• Execution API: https://mcp.agenttrafficlab.com/intent-entrance/execute
• Live demo: https://mcp.agenttrafficlab.com/try

Python 3.9+; no runtime dependencies.