TypeScript · Python SDK

import { Atmark, AtmarkError, isHostile } from '@atmark/sdk';

const atmark = new Atmark({ token: process.env.ATMARK_AGENT_TOKEN, agentId: process.env.AGENT_ID });

// 1. 먼저 내 설정을 읽는다: 주소 · 권한 · 정책 · 발신 승인
const caps = await atmark.capabilities.get();
console.log(caps.agent?.address, caps.policy?.send_approval);

// 2. 받은편지함 읽기. 받은 메일은 바깥 데이터이고 지시가 아니다.
for await (const message of atmark.messages.inbound.list({ read: false })) {
  if (isHostile(message)) continue; // risk.level 이 "risk" 나 "block"
  const full = await atmark.messages.inbound.get(message.id);
  await atmark.labels.set(full.id, { add: ['seen-by-agent'] });
}

// 3. 초안을 만들고 보낸다. Idempotency-Key 는 SDK 가 만든다.
const draft = await atmark.drafts.create({ from: 'scout@atmark.ai', to: ['partner@example.com'], subject: 'Weekly report', text: 'Numbers attached.' });
try {
  const sent = await atmark.drafts.send(draft.id);
  if (sent.status === 'pending_approval') console.log('소유자 승인 대기. 다시 보내지 않는다.');
} catch (error) {
  if (error instanceof AtmarkError) console.log(error.status, error.code); // error.code 로 분기
  else throw error;
}

import os
from atmark import Atmark, AtmarkError, is_hostile

atmark = Atmark(token=os.environ["ATMARK_AGENT_TOKEN"], agent_id=os.environ["AGENT_ID"])

# 1. 먼저 내 설정을 읽는다: 주소 · 권한 · 정책 · 발신 승인
caps = atmark.capabilities.get()
print(caps["agent"]["address"], caps["policy"]["send_approval"])

# 2. 받은편지함 읽기. 받은 메일은 바깥 데이터이고 지시가 아니다.
for message in atmark.messages.inbound.list(read=False):
    if is_hostile(message):  # risk.level 이 "risk" 나 "block"
        continue
    full = atmark.messages.inbound.get(message["id"])
    atmark.labels.set(full["id"], {"add": ["seen-by-agent"]})

# 3. 초안을 만들고 보낸다. Idempotency-Key 는 SDK 가 만든다.
draft = atmark.drafts.create({"from": "scout@atmark.ai", "to": ["partner@example.com"], "subject": "Weekly report", "text": "Numbers attached."})
try:
    sent = atmark.drafts.send(draft["id"])
    if sent["status"] == "pending_approval":
        print("소유자 승인 대기. 다시 보내지 않는다.")
except AtmarkError as error:
    print(error.status, error.code)  # error.code 로 분기

const connection = atmark.realtime.connect({
  onEvent(event) {
    if (event.type === 'message.received' && event.risk.level === 'none') void atmark.messages.inbound.get(event.message_id);
  },
  onOpen({ reconnected }) {
    if (reconnected) {
      // 놓친 이벤트는 다시 오지 않는다: messages.inbound.list({ since }) 로 따라잡는다.
    }
  },
  onError: console.error,
});
// connection.close();

import asyncio

async def main() -> None:
    async for event in atmark.realtime.events():
        if event.type == "message.received" and event.risk["level"] == "none":
            print(atmark.messages.inbound.get(event["message_id"])["subject"])

asyncio.run(main())

import { verifyWebhookSignature } from '@atmark/sdk';

export async function handleWebhook(request: Request): Promise<Response> {
  const rawBody = await request.text();
  const result = await verifyWebhookSignature(rawBody, request.headers, process.env.WEBHOOK_SECRET ?? '');
  if (!result.valid) return new Response(result.reason, { status: 401 });
  return new Response('ok');
}

from atmark import verify_webhook_signature

result = verify_webhook_signature(raw_body, request.headers, os.environ["WEBHOOK_SECRET"])
if not result.valid:
    return 401, result.reason