Choose your SDK
Same feature surface across both runtimes. Pick one — examples adapt automatically.
Installation
npm install @execlave/sdkZero runtime dependencies. Supports Node.js 18+ and modern bundlers for browser-side usage. TypeScript types included.
pip install execlave-sdkFor OpenTelemetry integration, install with extras:
pip install execlave-sdk[otel]Requires Python 3.9+. Core package has minimal dependencies.
Initialization
import { Execlave } from '@execlave/sdk'; const exe = new Execlave({ apiKey: process.env.EXECLAVE_API_KEY!, // required baseUrl: 'https://api.execlave.com', // default environment: 'production', // default asyncMode: true, // default: buffer traces batchSize: 100, // max traces per flush flushIntervalMs: 10_000, // 10s flush interval debug: false, // verbose logging enableControlChannel: true, // kill-switch polling pollIntervalMs: 15_000, // poll interval enableInjectionScan: true, // tags traces with injection score (does NOT block) enforcementOnOutage: 'fail_open', // fail_open | fail_closed onEnforcementBypassed: (e) => alert(e), // fires whenever an action ran UNGOVERNED policyCacheTtlMs: 60_000, // policy decision cache TTL});Create a single Execlave instance and share it across your app. The SDK manages connection pooling, batching, and background flushing internally.
from execlave import Execlave exe = Execlave( api_key=os.environ["EXECLAVE_API_KEY"], # required base_url="https://api.execlave.com", # default environment="production", # default privacy={ # PII scrubbing (off unless enabled=True) "enabled": True, "scrub_fields": ["input", "output"], # fields to scrub "hash_pii": True, # attach hashed PII summary to metadata }, enable_injection_scan=True, # tags traces with injection score (does NOT block) async_mode=True, # background flush thread; set False for sync mode="native", # native | otlp (transport) enforcement_on_outage="fail_open", # fail_open | fail_closed on_enforcement_bypassed=alert, # fires whenever an action ran UNGOVERNED policy_cache_ttl_seconds=60, # cache TTL (seconds))Methods
registerAgent
exe.registerAgent(config: AgentConfig): Promise<Agent>Register a new agent or update an existing one. Idempotent — safe to call on every app startup.
§ Parameters
agentIdstringUnique identifier for the agentnamestringDisplay nametype?stringchatbot | copilot | autonomous | workflow | data_processingplatform?stringcustom | openai | anthropic | langchainenvironment?'development' | 'staging' | 'production'Defaults to development. Pin to production when running in prod.description?stringAgent descriptionownerEmail?stringSurfaced in the dashboard and on policy violation alerts.allowedDataSources?string[]Governance allowlist of data sources the agent may read (e.g. ["s3://bucket-x", "postgres://orders"]). Enforced by data-access policies.allowedActions?string[]Governance allowlist of tool/action names this agent may invoke. Anything not on the list is blocked by default.requiresHumanApprovalFor?string[]Action names that always go to the human-approval queue regardless of policy outcome (e.g. ["wire_transfer", "delete_customer"]).tags?string[]Tags for filteringmetadata?Record<string, unknown>Free-form metadata stored on the agent record (cost-center IDs, owning team, ticket links).autonomyLevel?AutonomyLevelobserve | advise | act_with_approval | autonomous. Declares how much the agent may do unsupervised; drift detection can downgrade it later, and promotion to autonomous can be gated on a red-team score.§ Returns
Promise<Agent>await exe.registerAgent({ agentId: 'support-bot', name: 'Customer Support Bot', type: 'chatbot', platform: 'openai', environment: 'production', ownerEmail: 'support-team@example.com', allowedDataSources: ['postgres://customers'], allowedActions: ['search_kb', 'create_ticket'], requiresHumanApprovalFor: ['issue_refund'], tags: ['support', 'tier-1'], metadata: { costCenter: 'CS-101' },});enforcePolicy
exe.enforcePolicy(opts): Promise<EnforcementDecision>Synchronous pre-execution policy check. Call this BEFORE every LLM invocation. Throws PolicyBlockedError on a block-mode violation, AgentPausedError if the agent is kill-switched, and EnforcementUnavailableError only when enforcementOnOutage is 'fail_closed'. Tracing alone does NOT block — this method is the gate.
§ Parameters
agentIdstringAgent performing the callinputstringUser input being sent to the LLMenvironment?EnvironmentOverride the client environment for this checkmetadata?Record<string, unknown>Optional metadata for custom validatorsestimatedCost?numberEstimated call cost in USD (for budget policies)tools?string[]Tool names the agent intends to use (for action_approval policies)§ Returns
Promise<EnforcementDecision>. Throws PolicyBlockedError if blocked.import { AgentPausedError, PolicyBlockedError } from '@execlave/sdk'; try { await exe.enforcePolicy({ agentId: 'support-bot', input: userQuestion }); const answer = await llm.call(userQuestion);} catch (err) { if (err instanceof PolicyBlockedError) { // err.violations: list of { policyType, policyName, severity, message, enforcementMode } return 'Your input was blocked by our content policies.'; } if (err instanceof AgentPausedError) { return 'Service temporarily unavailable.'; } throw err;}startTrace
exe.startTrace(opts: TraceOptions): TraceStart a new trace to record an LLM interaction. Returns a chainable Trace object. Tracing is post-hoc — it does NOT block the LLM call. Pair with enforcePolicy() to actually block requests.
§ Parameters
agentIdstringAgent performing the tracesessionId?stringGroup traces into conversations§ Returns
Traceconst trace = exe.startTrace({ agentId: 'support-bot' });trace .setInput('How do I reset my password?') .setOutput('Go to Settings > Security...') .setModel('gpt-4') .setTokens(150, 320) .setCost(0.0045) .finish(); // Or finish with error:trace.finish('error', 'LLM call timed out');wrap
exe.wrap<T>(fn, opts): (input: string) => Promise<T>Wrap a function with automatic tracing. Input/output are captured, and the trace is finished on return or error.
§ Parameters
fnFunctionAsync function to wrapopts.agentIdstringAgent performing the call§ Returns
Wrapped function with same signatureconst tracedAnswer = exe.wrap( async (question: string) => { const res = await openai.chat.completions.create({ model: 'gpt-4', messages: [{ role: 'user', content: question }], }); return res.choices[0].message.content; }, { agentId: 'support-bot' }); // Usage — fully traced automaticallyconst answer = await tracedAnswer('How do I upgrade?');enforceToolOutput
exe.enforceToolOutput(opts): Promise<EnforceResult>Scan a tool's result BEFORE feeding it back to the model. This is what makes tool_output_scan preventive rather than detective — the framework adapters do not call it for you, so a result you never pass here cannot be blocked.
§ Parameters
agentIdstringAgent that invoked the tooltoolNamestringName of the tool that returnedoutputunknownThe raw tool result to scaninput?unknownArguments the tool was called with§ Returns
Promise<EnforceResult>const raw = await webSearch(query);await exe.enforceToolOutput({ agentId: 'bot', toolName: 'web_search', output: raw });// Throws PolicyBlockedError if the result carries denied PII or injection.messages.push({ role: 'tool', content: raw });verifyApproval
exe.verifyApproval(approvalId, actionContext): Promise<VerifyResult>Verify that a granted approval actually covers the action about to run. The certificate is bound to the action context, so a mismatch or a replay is rejected rather than silently honoured.
§ Parameters
approvalIdstringFrom a 202 enforce responseactionContextRecord<string, unknown>The action being performed — must match what was approved§ Returns
Promise<{ valid: boolean; reason?: string; certificate?: object }>const v = await exe.verifyApproval(approvalId, { input, environment: 'production' });if (!v.valid) throw new Error(`approval not usable: ${v.reason}`);checkAgentStatus
exe.checkAgentStatus(agentId?): Promise<string>Current lifecycle status of an agent — most importantly whether it has been paused by the kill switch.
§ Returns
Promise<'active' | 'paused' | 'unknown'>if ((await exe.checkAgentStatus('bot')) === 'paused') return;ping
exe.ping(): Promise<boolean>Liveness probe against the API. Useful at startup to fail fast on a misconfigured key or base URL.
§ Returns
Promise<boolean>if (!(await exe.ping())) console.warn('Execlave unreachable');flush
exe.flush(): Promise<void>Force-send buffered traces without shutting down. Use in short-lived processes (a serverless invocation) where the flush interval may never fire.
§ Returns
Promise<void>await exe.flush();checkUsage
exe.checkUsage(): Promise<UsageStatus>Current plan consumption. Lets you surface an approaching trace quota before ingestion starts failing.
§ Returns
Promise<UsageStatus>const usage = await exe.checkUsage();getAgentCredential
exe.getAgentCredential(agentId): Promise<AgentCredential>Fetch the agent's exe_agt_ credential, which proves agent identity on trace ingest and agent-to-agent calls. Cached in-process.
§ Returns
Promise<AgentCredential>const cred = await exe.getAgentCredential('bot');authorizeAgentCall
exe.authorizeAgentCall(opts: AuthorizeCallOptions): Promise<AuthorizeResult>Authorize one agent calling another (A2A). Requires FF_A2A_AUTH on the backend; records an a2a.authorize audit event.
§ Returns
Promise<AuthorizeResult>const decision = await exe.authorizeAgentCall({ callerAgentId: 'a', targetAgentId: 'b' });discoverAgents
exe.discoverAgents(capability?): Promise<DiscoveredAgent[]>List agents in the organization, optionally filtered by a declared capability.
§ Returns
Promise<DiscoveredAgent[]>const agents = await exe.discoverAgents('refunds');reportAgentMetadata
exe.reportAgentMetadata(opts: ReportAgentMetadataOptions): Promise<unknown>Report framework, model, and capability metadata for an agent. Feeds the agent passport and the inventory view.
§ Returns
Promise<unknown>await exe.reportAgentMetadata({ agentId: 'bot', framework: 'langchain' });toolDescriptor
exe.toolDescriptor(opts): ToolDescriptorBuild a hashed descriptor for one MCP tool. Synchronous — pair with reportToolBaseline to pin a supply-chain baseline.
§ Returns
{ server, tool, descriptorHash, description? }const d = exe.toolDescriptor({ server: 'files', tool: 'read', descriptor: schema });reportToolBaseline
exe.reportToolBaseline(opts: ReportToolBaselineOptions): Promise<unknown>Pin the current MCP tool descriptors as the agent's baseline. Later drift is detected against it by tool_integrity policies.
§ Returns
Promise<unknown>await exe.reportToolBaseline({ agentId: 'bot', tools: [d] });shutdown
exe.shutdown(): Promise<void>Flush all buffered traces and close connections. Call on process exit.
§ Returns
Promise<void>process.on('SIGTERM', async () => { await exe.shutdown(); process.exit(0);});register_agent
exe.register_agent(**kwargs) -> AgentRegister a new agent or update an existing one. Idempotent.
§ Parameters
agent_idstrUnique identifiernamestrDisplay nametypestrchatbot | copilot | autonomous | workflow | data_processingplatformstrcustom | openai | anthropic | langchain | ...descriptionstrAgent descriptionowner_emailstrSurfaced on dashboard and alertsallowed_data_sourceslist[str]Governance allowlist of data sources this agent may read.allowed_actionslist[str]Governance allowlist of tool/action names. Anything not on the list is blocked.requires_human_approval_forlist[str]Action names that always require human approval.tagslist[str]Tags for filteringmetadatadictFree-form metadata stored on the agent record.§ Returns
Agentagent = exe.register_agent( agent_id="support-bot", name="Customer Support Bot", type="chatbot", platform="langchain", owner_email="support-team@example.com", allowed_actions=["search_kb", "create_ticket"], requires_human_approval_for=["issue_refund"], tags=["support", "production"],)enforce_policy
exe.enforce_policy(agent_id, input, *, environment=None, metadata=None, estimated_cost=None, tools=None) -> dictSynchronous pre-execution policy check. Call this BEFORE every LLM invocation. Raises PolicyBlockedError on a block-mode violation, AgentPausedError if the agent is kill-switched, and EnforcementUnavailableError only when enforcement_on_outage='fail_closed'. Tracing alone does NOT block — this method is the gate.
§ Parameters
agent_idstrAgent performing the callinputstrUser input being sent to the LLMenvironmentstr | NoneOverride the client environment for this checkmetadatadict | NoneOptional metadata for custom validatorsestimated_costfloat | NoneEstimated call cost in USD (for budget policies)toolslist[str] | NoneTool names the agent intends to use (for action_approval policies)§ Returns
dict (decision payload). Raises PolicyBlockedError if blocked.from execlave import AgentPausedError, PolicyBlockedError try: exe.enforce_policy(agent_id="support-bot", input=user_question) response = llm.invoke(user_question)except PolicyBlockedError as e: # e.violations: list of {policyType, policyName, severity, message, enforcementMode} return "Your input was blocked by our content policies."except AgentPausedError: return "Service temporarily unavailable."@exe.trace
@exe.trace(*, agent_id=None, session_id=None, user_id=None, metadata=None, tags=None, environment=None, parent_trace_id=None, span_type=None)Decorator / context manager that automatically traces function calls. Arguments become the trace input, return value becomes output.
§ Parameters
agent_idstr | NoneOverride the default agent IDsession_idstr | NoneGroup traces into a sessionuser_idstr | NoneAttribute the trace to an end usermetadatadict | NoneArbitrary metadata attached to the tracetagslist[str] | NoneTags for filtering tracesenvironmentstr | NoneOverride the client environment for this traceparent_trace_idstr | NoneLink this span to a parent trace (nesting)span_typestr | Noneroot | agent | llm_call | tool_call | retrieval | middleware | custom§ Returns
Decorated function (or TraceContext when used as a context manager)@exe.tracedef answer(question: str) -> str: return llm.invoke(question) # Session + user attribution:@exe.trace(session_id="sess_123", user_id="user_42")def handle(query: str) -> str: return db.execute(query)start_trace
exe.start_trace(agent_id: str) -> TraceContextContext manager for manual tracing with full control over trace fields.
§ Parameters
agent_idstrAgent performing the trace§ Returns
TraceContext (context manager)with exe.start_trace(agent_id="support-bot") as trace: trace.set_input(user_question) response = llm.invoke(user_question) trace.set_output(response) trace.set_model("gpt-4") trace.set_tokens(input=150, output=320) trace.set_cost(0.0045) trace.add_metadata({"intent": "password_reset"})enforce_tool_output
exe.enforce_tool_output(agent_id, tool_name, output, *, input=None) -> dictScan a tool's result BEFORE feeding it back to the model. This is what makes tool_output_scan preventive rather than detective — the framework adapters do not call it for you, so a result you never pass here cannot be blocked.
§ Parameters
agent_idstrAgent that invoked the tooltool_namestrName of the tool that returnedoutputAnyThe raw tool result to scaninputAny | NoneArguments the tool was called with§ Returns
dictraw = web_search(query)exe.enforce_tool_output(agent_id="bot", tool_name="web_search", output=raw)# Raises PolicyBlockedError if the result carries denied PII or injection.messages.append({"role": "tool", "content": raw})verify_approval
exe.verify_approval(approval_id, action_context) -> dictVerify that a granted approval actually covers the action about to run. The certificate is bound to the action context, so a mismatch or a replay is rejected rather than silently honoured.
§ Returns
dict — { valid, reason?, certificate? }v = exe.verify_approval(approval_id, {"input": text, "environment": "production"})if not v["valid"]: raise RuntimeError(f"approval not usable: {v.get('reason')}")check_agent_status
exe.check_agent_status(agent_id=None) -> strCurrent lifecycle status of an agent — most importantly whether it has been paused by the kill switch.
§ Returns
str — 'active' | 'paused' | 'unknown'if exe.check_agent_status("bot") == "paused": returnping
exe.ping() -> boolLiveness probe against the API. Useful at startup to fail fast on a misconfigured key or base URL.
§ Returns
boolif not exe.ping(): logger.warning("Execlave unreachable")flush
exe.flush() -> NoneForce-send buffered traces without shutting down. Use in short-lived processes (a serverless invocation) where the flush interval may never fire.
§ Returns
Noneexe.flush()check_usage
exe.check_usage() -> dictCurrent plan consumption. Lets you surface an approaching trace quota before ingestion starts failing.
§ Returns
dictusage = exe.check_usage()get_agent_credential
exe.get_agent_credential(agent_id) -> dictFetch the agent's exe_agt_ credential, which proves agent identity on trace ingest and agent-to-agent calls. Cached in-process.
§ Returns
dictcred = exe.get_agent_credential("bot")authorize_agent_call
exe.authorize_agent_call(...) -> dictAuthorize one agent calling another (A2A). Requires FF_A2A_AUTH on the backend; records an a2a.authorize audit event.
§ Returns
dictdecision = exe.authorize_agent_call(caller_agent_id="a", target_agent_id="b")discover_agents
exe.discover_agents(capability=None) -> list[dict]List agents in the organization, optionally filtered by a declared capability.
§ Returns
list[dict]agents = exe.discover_agents("refunds")report_agent_metadata
exe.report_agent_metadata(...) -> AnyReport framework, model, and capability metadata for an agent. Feeds the agent passport and the inventory view.
§ Returns
Anyexe.report_agent_metadata(agent_id="bot", framework="langchain")tool_descriptor
exe.tool_descriptor(server, tool, descriptor, description=None) -> dictBuild a hashed descriptor for one MCP tool. Synchronous — pair with report_tool_baseline to pin a supply-chain baseline.
§ Returns
dictd = exe.tool_descriptor(server="files", tool="read", descriptor=schema)report_tool_baseline
exe.report_tool_baseline(...) -> AnyPin the current MCP tool descriptors as the agent's baseline. Later drift is detected against it by tool_integrity policies.
§ Returns
Anyexe.report_tool_baseline(agent_id="bot", tools=[d])Error types
Thrown when a trace is started for an agent that has been paused via the kill switch. Your app should catch this and return a graceful fallback to the user.
Thrown when a pre-execution policy check blocks the request. Contains the violated policy name and enforcement mode.
| Error | Thrown when |
|---|---|
EnforcementUnavailableError | Enforcement could not be reached AND enforcementOnOutage is fail_closed. Under the default fail_open this is NOT thrown — the call returns allowed:true instead, which is what onEnforcementBypassed reports. |
PolicyDeniedError | A policy denied the action outright, as distinct from a block-mode violation. |
ApprovalTimeoutError | An approval was required and no decision arrived before the timeout elapsed. |
ApprovalVerificationError | verifyApproval() could not confirm the approval covers this action. |
CertificateMismatchError | The approval certificate does not bind to the action being performed — a swapped or replayed certificate. |
ToolIntegrityError | A presented MCP tool descriptor diverges from the pinned baseline. |
ValidatorDeniedError | A custom (BYOV) validator returned a deny decision. |
PlanLimitExceededError | A plan limit was hit AND planLimitBehavior is fail_closed. Under the default fail_open execution continues unmonitored and the bypass is reported instead. |
QuotaExceededError | Trace ingestion quota for the billing period is exhausted. |
EnforcementHaltError | An adapter halted the run because enforcement could not be completed safely. |
MetadataContractError | Sealed action metadata was altered between enforcement and execution. |
ExeclaveAuthError | The API key or token was rejected. |
ExeclaveError | Base class for every error above — catch this to handle any SDK failure generically. |
Python raises the same set under snake_case module paths (execlave.errors), with identical class names.
Privacy & PII scrubbing
The TypeScript SDK can score every input for prompt-injection likelihood and tag the resulting trace with that score. This is metadata only — it does not block the LLM call. To actually block injection attempts, create an injection_scan policy in block mode and call exe.enforcePolicy() before your LLM invocation. Server-side PII scrubbing is performed by the processing service on ingestion.
const exe = new Execlave({ apiKey: process.env.EXECLAVE_API_KEY!, enableInjectionScan: true, // tags traces with injection score (does NOT block)}); // To block: create a block-mode injection_scan policy in the dashboard,// then gate every LLM call with enforcePolicy().await exe.enforcePolicy({ agentId: 'support-bot', input: userQuestion });The Python SDK can scrub PII before traces leave your application, and can score every input for prompt-injection likelihood (metadata only — it does not block the LLM call). To actually block injection attempts, create an injection_scan policy in block mode and call exe.enforce_policy() before your LLM invocation.
exe = Execlave( api_key=os.environ["EXECLAVE_API_KEY"], privacy={ "enabled": True, "scrub_fields": ["input", "output"], # which trace fields to scrub "hash_pii": False, # mask vs hash matched values }, enable_injection_scan=True, # tags traces with injection score (does NOT block)) # To block: create a block-mode injection_scan policy in the dashboard,# then gate every LLM call with enforce_policy().exe.enforce_policy(agent_id="support-bot", input=user_question)PII is scrubbed client-side before transmission, so sensitive data never reaches the Execlave server.
OpenTelemetry integration
Switch the SDK into OTLP transport by setting mode: 'otlp' and pointing otlpEndpoint at the base URL of your OTLP collector (the SDK appends /v1/traces automatically). Use an OTLP-capable collector — do not point this at the Execlave REST API.
import { Execlave } from '@execlave/sdk'; const exe = new Execlave({ apiKey: process.env.EXECLAVE_API_KEY!, mode: 'otlp', // Base URL of your OTLP collector — exporter appends /v1/traces. // e.g. http://localhost:4317 or your managed collector's base URL. otlpEndpoint: process.env.OTLP_ENDPOINT!,}); // Traces emitted by the SDK are exported via OTLP instead of the// native REST ingest, so they flow through your existing collector// pipeline alongside the rest of your telemetry.Switch the SDK into OTLP transport by passing mode="otlp" and an otlp_endpoint pointing at the base URL of your OTLP collector (the SDK appends /v1/traces automatically). Use an OTLP-capable collector — do not point this at the Execlave REST API.
import osfrom execlave import Execlave exe = Execlave( api_key=os.environ["EXECLAVE_API_KEY"], mode="otlp", # Base URL of your OTLP collector — exporter appends /v1/traces. # e.g. http://localhost:4317 or your managed collector's base URL. otlp_endpoint=os.environ["OTLP_ENDPOINT"],) # Traces emitted by the SDK are exported via OTLP instead of the# native REST ingest, so they flow through your existing collector# pipeline alongside the rest of your telemetry.