The full agent loop
The complete sequence — create envelope from template, wait for signature via SSE, download signed PDF.
import * as fs from 'fs';
const ATLAS_KEY = process.env.ATLAS_API_KEY!;
const BASE = 'https://atlaswork.ai';
// ── 1. Create envelope from a saved PDF template ──────────────
const { envelope_id, sign_url } = await fetch(`${BASE}/api/envelope`, {
method: 'POST',
headers: {
'Authorization': `Bearer ${ATLAS_KEY}`,
'Content-Type': 'application/json',
},
body: JSON.stringify({
template_id: 'tmpl_abc123', // from Atlas dashboard
signer_email: 'alice@company.com',
agent_identity: 'my-agent/v1',
prefill: {
'Client Name': 'Acme Corp',
'Deal Amount': '$125,000',
'Effective Date': 'April 1, 2026',
},
}),
}).then(r => r.json());
console.log('Signing link:', sign_url);
// ── 2. Stream SSE events until terminal state ─────────────────
const stream = await fetch(`${BASE}/api/envelope/${envelope_id}/events`, {
headers: { 'Authorization': `Bearer ${ATLAS_KEY}` },
});
outer: for await (const chunk of stream.body as AsyncIterable<Uint8Array>) {
for (const line of new TextDecoder().decode(chunk).split('\n')) {
if (!line.startsWith('data: ')) continue;
const { status, envelope } = JSON.parse(line.slice(6));
console.log('Status:', status);
if (status === 'signed') {
console.log('Document hash:', envelope.document_hash);
break outer;
}
if (status === 'declined' || status === 'voided') {
throw new Error(`Signing ${status}`);
}
}
}
// ── 3. Download the signed PDF ────────────────────────────────
const pdf = await fetch(`${BASE}/api/envelope/${envelope_id}/pdf`, {
headers: { 'Authorization': `Bearer ${ATLAS_KEY}` },
});
fs.writeFileSync('signed.pdf', Buffer.from(await pdf.arrayBuffer()));
console.log('Saved to signed.pdf');LangChain (Python)
Two tools registered with a tool-calling agent: send a signing request and check envelope status.
import os, json, httpx
from langchain.tools import tool
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate
ATLAS_KEY = os.environ["ATLAS_API_KEY"]
BASE = "https://atlaswork.ai"
@tool
def send_signature_request(
template_id: str,
signer_email: str,
prefill: dict = {},
) -> str:
"""Send a document for signing using a saved Atlas PDF template.
Args:
template_id: The Atlas template ID (e.g. 'tmpl_abc123')
signer_email: Email address of the signer
prefill: Optional key-value pairs to pre-fill text fields
Returns:
Signing URL and envelope ID
"""
res = httpx.post(
f"{BASE}/api/envelope",
headers={"Authorization": f"Bearer {ATLAS_KEY}"},
json={
"template_id": template_id,
"signer_email": signer_email,
"agent_identity": "langchain-agent/v1",
"prefill": prefill,
},
timeout=30,
)
res.raise_for_status()
data = res.json()
return f"Sent. Sign URL: {data['sign_url']} | Envelope ID: {data['envelope_id']}"
@tool
def get_envelope_status(envelope_id: str) -> str:
"""Check the current status of a signing envelope.
Args:
envelope_id: The envelope ID to check
Returns:
Current status (pending/signed/declined/voided) and document hash if signed
"""
res = httpx.get(
f"{BASE}/api/envelope/{envelope_id}",
headers={"Authorization": f"Bearer {ATLAS_KEY}"},
timeout=30,
)
res.raise_for_status()
env = res.json()["envelope"]
result = f"Status: {env['status']}"
if env.get("document_hash"):
result += f" | Hash: {env['document_hash']}"
return result
# Wire up
llm = ChatOpenAI(model="gpt-4o", temperature=0)
tools = [send_signature_request, get_envelope_status]
prompt = ChatPromptTemplate.from_messages([
("system", "You are a contract assistant. Use Atlas tools to send and track signing requests."),
("human", "{input}"),
("placeholder", "{agent_scratchpad}"),
])
agent = create_tool_calling_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
result = executor.invoke({
"input": "Send the NDA template tmpl_abc123 to alice@company.com with Client Name = Acme Corp"
})
print(result["output"])OpenAI Agents SDK
Async function tools with SSE streaming — the agent blocks until the document reaches a terminal state.
import os, json, httpx, asyncio
from agents import Agent, Runner, function_tool
ATLAS_KEY = os.environ["ATLAS_API_KEY"]
BASE = "https://atlaswork.ai"
@function_tool
async def send_envelope(
template_id: str,
signer_email: str,
prefill: dict | None = None,
) -> str:
"""Send a document for signing using a saved Atlas PDF template."""
async with httpx.AsyncClient() as client:
res = await client.post(
f"{BASE}/api/envelope",
headers={"Authorization": f"Bearer {ATLAS_KEY}"},
json={
"template_id": template_id,
"signer_email": signer_email,
"agent_identity": "openai-agents-sdk/v1",
"prefill": prefill or {},
},
timeout=30,
)
res.raise_for_status()
data = res.json()
return f"Sent. URL: {data['sign_url']} | ID: {data['envelope_id']}"
@function_tool
async def wait_for_signature(envelope_id: str) -> str:
"""Block until the envelope is signed, declined, or voided via SSE."""
async with httpx.AsyncClient(timeout=600) as client:
async with client.stream(
"GET",
f"{BASE}/api/envelope/{envelope_id}/events",
headers={"Authorization": f"Bearer {ATLAS_KEY}"},
) as stream:
async for line in stream.aiter_lines():
if not line.startswith("data: "):
continue
event = json.loads(line[6:])
if event["status"] in ("signed", "declined", "voided"):
env = event["envelope"]
return f"Status: {event['status']} | Hash: {env.get('document_hash', 'N/A')}"
return "Timed out"
contract_agent = Agent(
name="ContractAgent",
instructions="You help send and track contract signings using Atlas ESIGN.",
tools=[send_envelope, wait_for_signature],
)
async def main():
result = await Runner.run(
contract_agent,
"Send template tmpl_abc123 to alice@company.com then wait for her signature.",
)
print(result.final_output)
asyncio.run(main())LangGraph
A three-node graph: create envelope → poll status → download PDF, with a conditional edge that loops until a terminal state.
import os, json, httpx
from typing import TypedDict, Literal
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
ATLAS_KEY = os.environ["ATLAS_API_KEY"]
BASE = "https://atlaswork.ai"
class SigningState(TypedDict):
template_id: str
signer_email: str
prefill: dict
envelope_id: str | None
sign_url: str | None
status: str | None
document_hash: str | None
def create_envelope(state: SigningState) -> SigningState:
"""Node: create the envelope and return sign URL."""
res = httpx.post(
f"{BASE}/api/envelope",
headers={"Authorization": f"Bearer {ATLAS_KEY}"},
json={
"template_id": state["template_id"],
"signer_email": state["signer_email"],
"agent_identity": "langgraph/v1",
"prefill": state.get("prefill", {}),
},
timeout=30,
)
res.raise_for_status()
data = res.json()
return {**state, "envelope_id": data["envelope_id"], "sign_url": data["sign_url"]}
def poll_status(state: SigningState) -> SigningState:
"""Node: check current envelope status (non-blocking poll)."""
res = httpx.get(
f"{BASE}/api/envelope/{state['envelope_id']}",
headers={"Authorization": f"Bearer {ATLAS_KEY}"},
timeout=30,
)
env = res.json()["envelope"]
return {**state, "status": env["status"], "document_hash": env.get("document_hash")}
def download_pdf(state: SigningState) -> SigningState:
"""Node: download the signed PDF."""
res = httpx.get(
f"{BASE}/api/envelope/{state['envelope_id']}/pdf",
headers={"Authorization": f"Bearer {ATLAS_KEY}"},
follow_redirects=True,
timeout=30,
)
with open("signed.pdf", "wb") as f:
f.write(res.content)
print("Signed PDF saved to signed.pdf")
return state
def route_after_poll(state: SigningState) -> Literal["download_pdf", "poll_status", "end"]:
"""Conditional edge: route based on envelope status."""
if state["status"] == "signed":
return "download_pdf"
if state["status"] in ("declined", "voided"):
return "end"
return "poll_status" # keep polling
# Build graph
graph = StateGraph(SigningState)
graph.add_node("create_envelope", create_envelope)
graph.add_node("poll_status", poll_status)
graph.add_node("download_pdf", download_pdf)
graph.set_entry_point("create_envelope")
graph.add_edge("create_envelope", "poll_status")
graph.add_conditional_edges("poll_status", route_after_poll, {
"download_pdf": "download_pdf",
"poll_status": "poll_status",
"end": END,
})
graph.add_edge("download_pdf", END)
app = graph.compile()
result = app.invoke({
"template_id": "tmpl_abc123",
"signer_email": "alice@company.com",
"prefill": {"Client Name": "Acme Corp"},
})
print("Final status:", result["status"])Vercel AI SDK
Drop-in route handler with Zod-typed tools for Next.js App Router streaming responses.
// app/api/agent/route.ts
import { openai } from '@ai-sdk/openai';
import { streamText, tool } from 'ai';
import { z } from 'zod';
const ATLAS_KEY = process.env.ATLAS_API_KEY!;
const BASE = 'https://atlaswork.ai';
export async function POST(req: Request) {
const { messages } = await req.json();
const result = streamText({
model: openai('gpt-4o'),
system: 'You are a contract assistant. Use Atlas tools to send and track document signings.',
messages,
tools: {
sendEnvelope: tool({
description: 'Send a document for signing using a saved Atlas PDF template.',
parameters: z.object({
template_id: z.string().describe('Atlas template ID'),
signer_email: z.string().email().describe('Signer email address'),
prefill: z.record(z.string()).optional().describe('Field label → value map'),
}),
execute: async ({ template_id, signer_email, prefill }) => {
const res = await fetch(`${BASE}/api/envelope`, {
method: 'POST',
headers: {
'Authorization': `Bearer ${ATLAS_KEY}`,
'Content-Type': 'application/json',
},
body: JSON.stringify({
template_id,
signer_email,
agent_identity: 'vercel-ai-sdk/v1',
prefill: prefill ?? {},
}),
});
const data = await res.json();
return { sign_url: data.sign_url, envelope_id: data.envelope_id };
},
}),
getEnvelopeStatus: tool({
description: 'Check the status of a signing envelope.',
parameters: z.object({
envelope_id: z.string().describe('The envelope ID to check'),
}),
execute: async ({ envelope_id }) => {
const res = await fetch(`${BASE}/api/envelope/${envelope_id}`, {
headers: { 'Authorization': `Bearer ${ATLAS_KEY}` },
});
const { envelope } = await res.json();
return {
status: envelope.status,
document_hash: envelope.document_hash ?? null,
};
},
}),
},
});
return result.toDataStreamResponse();
}AutoGen
Register Atlas functions on an AssistantAgent and UserProxyAgent pair usingautogen.register_function.
import os, json, httpx
import autogen
ATLAS_KEY = os.environ["ATLAS_API_KEY"]
BASE = "https://atlaswork.ai"
def send_envelope(template_id: str, signer_email: str, prefill: dict = {}) -> str:
"""Send a signing request using an Atlas PDF template."""
res = httpx.post(
f"{BASE}/api/envelope",
headers={"Authorization": f"Bearer {ATLAS_KEY}"},
json={
"template_id": template_id,
"signer_email": signer_email,
"agent_identity": "autogen/v1",
"prefill": prefill,
},
timeout=30,
)
res.raise_for_status()
data = res.json()
return f"Sent. Sign URL: {data['sign_url']} | Envelope ID: {data['envelope_id']}"
def get_envelope_status(envelope_id: str) -> str:
"""Get current status of a signing envelope."""
res = httpx.get(
f"{BASE}/api/envelope/{envelope_id}",
headers={"Authorization": f"Bearer {ATLAS_KEY}"},
timeout=30,
)
env = res.json()["envelope"]
return f"Status: {env['status']}" + (f" | Hash: {env['document_hash']}" if env.get("document_hash") else "")
config_list = [{"model": "gpt-4o", "api_key": os.environ["OPENAI_API_KEY"]}]
assistant = autogen.AssistantAgent(
name="ContractAssistant",
system_message="You manage document signing using Atlas ESIGN API tools.",
llm_config={"config_list": config_list},
)
user_proxy = autogen.UserProxyAgent(
name="User",
human_input_mode="NEVER",
max_consecutive_auto_reply=10,
code_execution_config=False,
)
# Register the tools
autogen.register_function(
send_envelope,
caller=assistant,
executor=user_proxy,
name="send_envelope",
description="Send a document for signing using an Atlas PDF template",
)
autogen.register_function(
get_envelope_status,
caller=assistant,
executor=user_proxy,
name="get_envelope_status",
description="Check the signing status of an envelope",
)
user_proxy.initiate_chat(
assistant,
message="Send the NDA template tmpl_abc123 to alice@company.com with Client Name = Acme Corp. Then check if she has signed.",
)Pydantic AI
Typed async tools with a dependency-injected HTTP client passed throughAtlasDeps.
import os, json, httpx, asyncio
from dataclasses import dataclass
from pydantic_ai import Agent
from pydantic_ai.models.openai import OpenAIModel
ATLAS_KEY = os.environ["ATLAS_API_KEY"]
BASE = "https://atlaswork.ai"
@dataclass
class AtlasDeps:
api_key: str
http: httpx.AsyncClient
model = OpenAIModel("gpt-4o")
agent = Agent(
model,
deps_type=AtlasDeps,
system_prompt="You are a contract assistant. Use Atlas tools to send and track document signings.",
)
@agent.tool
async def send_envelope(
ctx,
template_id: str,
signer_email: str,
client_name: str = "",
deal_amount: str = "",
) -> str:
"""Send a document for signing using a saved Atlas PDF template."""
prefill = {}
if client_name: prefill["Client Name"] = client_name
if deal_amount: prefill["Deal Amount"] = deal_amount
res = await ctx.deps.http.post(
f"{BASE}/api/envelope",
headers={"Authorization": f"Bearer {ctx.deps.api_key}"},
json={
"template_id": template_id,
"signer_email": signer_email,
"agent_identity": "pydantic-ai/v1",
"prefill": prefill,
},
timeout=30,
)
res.raise_for_status()
data = res.json()
return f"Sent. URL: {data['sign_url']} | ID: {data['envelope_id']}"
@agent.tool
async def get_envelope_status(ctx, envelope_id: str) -> str:
"""Check the current status of a signing envelope."""
res = await ctx.deps.http.get(
f"{BASE}/api/envelope/{envelope_id}",
headers={"Authorization": f"Bearer {ctx.deps.api_key}"},
timeout=30,
)
env = res.json()["envelope"]
return f"Status: {env['status']}" + (f" | Hash: {env['document_hash']}" if env.get("document_hash") else "")
async def main():
async with httpx.AsyncClient() as http:
deps = AtlasDeps(api_key=ATLAS_KEY, http=http)
result = await agent.run(
"Send template tmpl_abc123 to alice@company.com with Client Name = Acme Corp",
deps=deps,
)
print(result.data)
asyncio.run(main())TypeScript SDK
A typed TypeScript SDK wraps the REST API for the most common operations.
npm install @atlasapi/sdk
import { Atlas } from '@atlasapi/sdk';
const atlas = new Atlas({ apiKey: process.env.ATLAS_API_KEY! });
// Create an envelope
const { sign_url, envelope_id } = await atlas.envelopes.create({
template_id: 'tmpl_abc123',
signer_email: 'alice@company.com',
prefill: { 'Client Name': 'Acme Corp' },
});
// Wait for signature (SSE — resolves when terminal state reached)
const { status, document_hash } = await atlas.envelopes.waitForSignature(envelope_id);
// Download the signed PDF as a Buffer
const pdfBuffer = await atlas.envelopes.downloadPdf(envelope_id);Environment setup
Set these variables before running any of the examples above.
# .env
ATLAS_API_KEY=atlas_live_xxxxxxxxxxxxxxxxxxxx # live key — legally binding
ATLAS_TEST_KEY=test_xxxxxxxxxxxxxxxxxxxx # test key — NOT legally binding, shows TEST banner
ATLAS_WEBHOOK_SECRET=your_live_api_key # same as live key — used to verify webhook HMAC
# Verify your setup with the test key:
curl https://atlaswork.ai/api/envelope \
-H "Authorization: Bearer $ATLAS_TEST_KEY" \
-H "Content-Type: application/json" \
-d '{"base_template":"Test contract","proposed_markdown":"Test contract v2","signer_email":"you@example.com"}'
# { "success": true, "sign_url": "/sign/...", "envelope_id": "..." }
# The sign page shows a TEST MODE banner — not legally binding.