Cách Mình Phát Triển Agents Trên Lark

User Message → Lark Platform → Your Webhook → AI Agent → Lark API → Response
     ↑                                            ↓
     └────────────────────────────────────────────┘

https://your-domain.com/oauth/callback

{
  "type": "url_verification",
  "challenge": "random-string-here"
}

{
  "challenge": "random-string-here"
}

@app.post("/webhook/event")
async def webhook(request: Request):
    body = await request.json()

    # Xử lý challenge verification
    if body.get("type") == "url_verification":
        return {"challenge": body.get("challenge")}

    # Xử lý các events khác...

{
  "challenge": "random-string-here"
}

# Format Event Subscription
if body.get("type") == "url_verification":
    return {"challenge": body.get("challenge")}

# Format Callback Configuration (không có field type)
if "challenge" in body and "type" not in body and "header" not in body:
    return {"challenge": body.get("challenge")}

import base64
import hashlib
from Crypto.Cipher import AES

def decrypt_lark_event(encrypt_key: str, encrypted_data: str) -> dict:
    """Giải mã Lark encrypted event bằng AES-256-CBC."""
    # Tạo key bằng SHA256
    key = hashlib.sha256(encrypt_key.encode()).digest()

    # Decode base64
    encrypted_bytes = base64.b64decode(encrypted_data)

    # Tách IV (16 bytes đầu) và ciphertext
    iv = encrypted_bytes[:16]
    ciphertext = encrypted_bytes[16:]

    # Giải mã
    cipher = AES.new(key, AES.MODE_CBC, iv)
    decrypted = cipher.decrypt(ciphertext)

    # Xóa PKCS7 padding
    padding_len = decrypted[-1]
    decrypted = decrypted[:-padding_len]

    return json.loads(decrypted.decode('utf-8'))

{
  "type": "event_callback",
  "uuid": "unique-event-id",
  "event": {
    "type": "message",
    "text": "Hello!",
    "open_id": "ou_xxx",
    "open_message_id": "om_xxx",
    "chat_type": "p2p"
  }
}

{
  "header": {
    "event_id": "unique-event-id",
    "event_type": "im.message.receive_v1"
  },
  "event": {
    "sender": {
      "sender_id": {
        "open_id": "ou_xxx",
        "user_id": "xxx"
      }
    },
    "message": {
      "message_id": "om_xxx",
      "content": "{\"text\": \"Hello!\"}",
      "chat_type": "p2p"
    }
  }
}

from collections import OrderedDict
from datetime import datetime, timedelta

class EventCache:
    def __init__(self, max_size=1000, ttl_minutes=10):
        self._cache = OrderedDict()
        self._max_size = max_size
        self._ttl = timedelta(minutes=ttl_minutes)

    def is_processed(self, event_id: str) -> bool:
        if event_id in self._cache:
            if datetime.now() - self._cache[event_id] < self._ttl:
                return True
        return False

    def mark_processed(self, event_id: str):
        self._cache[event_id] = datetime.now()
        # Dọn entries cũ
        if len(self._cache) > self._max_size:
            self._cache.popitem(last=False)

1. User yêu cầu thao tác calendar/task
2. Bot phát hiện không có token hợp lệ
3. Bot gửi OAuth URL cho user
4. User click link, xác thực trong browser
5. Lark redirect đến /oauth/callback với code
6. Server đổi code lấy tokens
7. Tokens lưu vào database
8. User dùng được calendar/task

from urllib.parse import urlencode

def generate_oauth_url(app_id: str, redirect_uri: str, state: str) -> str:
    """Tạo Lark OAuth authorization URL."""
    scopes = " ".join([
        "offline_access",
        "calendar:calendar",
        "task:task:write",
        "task:tasklist:write",
        "task:section:read",
        "contact:user.id:readonly",
        "contact:user.base:readonly",
    ])

    params = {
        "app_id": app_id,
        "redirect_uri": redirect_uri,
        "state": state,  # Dùng open_id của user để tracking
        "scope": scopes,
    }

    return f"https://open.larksuite.com/open-apis/authen/v1/authorize?{urlencode(params)}"

https://your-domain.com/oauth/callback?code=xxx&state=xxx

async def exchange_code(code: str, app_id: str, app_secret: str, redirect_uri: str):
    """Đổi authorization code lấy tokens."""
    url = "https://open.larksuite.com/open-apis/authen/v2/oauth/token"

    payload = {
        "grant_type": "authorization_code",
        "client_id": app_id,
        "client_secret": app_secret,
        "code": code,
        "redirect_uri": redirect_uri,
    }

    async with httpx.AsyncClient() as client:
        response = await client.post(url, json=payload)
        data = response.json()

        if data.get("code") not in [0, "0"]:
            raise Exception(f"OAuth thất bại: {data}")

        return {
            "access_token": data["access_token"],
            "refresh_token": data["refresh_token"],
            "expires_in": data["expires_in"],  # Thường là 7200 (2 giờ)
        }

async def get_user_info(access_token: str):
    """Lấy thông tin user bằng access token."""
    url = "https://open.larksuite.com/open-apis/authen/v1/user_info"

    async with httpx.AsyncClient() as client:
        response = await client.get(
            url,
            headers={"Authorization": f"Bearer {access_token}"}
        )
        data = response.json()

        if data.get("code") != 0:
            raise Exception(f"Không lấy được thông tin user: {data}")

        return data.get("data")  # Chứa open_id, name, email, v.v.

async def refresh_token(refresh_token: str, app_id: str, app_secret: str):
    """Refresh access token."""
    url = "https://open.larksuite.com/open-apis/authen/v2/oauth/token"

    payload = {
        "grant_type": "refresh_token",
        "client_id": app_id,
        "client_secret": app_secret,
        "refresh_token": refresh_token,
    }

    async with httpx.AsyncClient() as client:
        response = await client.post(url, json=payload)
        data = response.json()

        if data.get("code") not in [0, "0"]:
            raise Exception(f"Refresh thất bại: {data}")

        return {
            "access_token": data["access_token"],
            "refresh_token": data["refresh_token"],
            "expires_in": data["expires_in"],
        }

async def get_tenant_token(app_id: str, app_secret: str) -> str:
    """Lấy tenant access token cho thao tác cấp app."""
    url = "https://open.larksuite.com/open-apis/auth/v3/tenant_access_token/internal"

    async with httpx.AsyncClient() as client:
        response = await client.post(url, json={
            "app_id": app_id,
            "app_secret": app_secret,
        })
        data = response.json()

        if data.get("code") != 0:
            raise Exception(f"Không lấy được tenant token: {data}")

        return data["tenant_access_token"]

async def send_message(message_id: str, content: str, tenant_token: str):
    """Reply tin nhắn."""
    url = f"https://open.larksuite.com/open-apis/im/v1/messages/{message_id}/reply"

    payload = {
        "content": json.dumps({"text": content}),
        "msg_type": "text",
    }

    async with httpx.AsyncClient() as client:
        response = await client.post(
            url,
            json=payload,
            headers={"Authorization": f"Bearer {tenant_token}"}
        )
        return response.json()

GET /calendar/v4/calendars
Authorization: Bearer {user_access_token}

{
  "code": 0,
  "data": {
    "calendar_list": [
      {
        "calendar_id": "xxx",
        "summary": "Primary",
        "type": "primary"
      }
    ]
  }
}

GET /calendar/v4/calendars/{calendar_id}/events
  ?start_time={timestamp_seconds}
  &end_time={timestamp_seconds}
Authorization: Bearer {user_access_token}

POST /calendar/v4/calendars/{calendar_id}/events
Authorization: Bearer {user_access_token}
Content-Type: application/json

{
  "summary": "Meeting Title",
  "start_time": {
    "timestamp": "1706140800",  // SECONDS
    "timezone": "Asia/Ho_Chi_Minh"
  },
  "end_time": {
    "timestamp": "1706144400",
    "timezone": "Asia/Ho_Chi_Minh"
  },
  "description": "Optional description",
  "location": {"name": "Meeting Room A"}
}

GET /task/v2/tasks?page_size=50
Authorization: Bearer {user_access_token}

POST /task/v2/tasks
Authorization: Bearer {user_access_token}
Content-Type: application/json

{
  "summary": "Task title",
  "due": {
    "timestamp": "1706140800000",  // MILLISECONDS
    "is_all_day": false
  }
}

POST /task/v2/tasks/{task_id}/complete
Authorization: Bearer {user_access_token}

GET /contact/v3/users/{user_id}?user_id_type=open_id
Authorization: Bearer {user_access_token}

lark-ai-agent/
├── agent.py                  # Entry point, FastAPI app
├── src/
│   ├── config.py             # Environment variables
│   ├── agent_config.py       # Setup AI agent
│   ├── lark_webhook.py       # Webhook handlers
│   ├── lark_api.py           # Tenant token API (replies)
│   ├── utils.py              # Utilities (encryption, v.v.)
│   │
│   ├── lark/                 # Lark API clients
│   │   ├── auth.py           # OAuth utilities
│   │   ├── client.py         # Base HTTP client
│   │   ├── calendar.py       # Calendar API
│   │   ├── tasks.py          # Task API
│   │   └── contact.py        # Contact API
│   │
│   ├── tools/                # AI agent tools
│   │   └── lark_toolkit.py   # Toolkit với tất cả tools
│   │
│   └── db/                   # Database
│       └── tokens.py         # Quản lý tokens
│
├── migrations/               # Database migrations
├── requirements.txt
├── Dockerfile
├── docker-compose.yaml
└── .env.example

async def handle_webhook(request: Request):
    body = await request.json()

    # 1. Giải mã nếu có encryption
    if "encrypt" in body:
        body = decrypt_lark_event(ENCRYPT_KEY, body["encrypt"])

    # 2. Challenge verification
    if body.get("type") == "url_verification":
        return {"challenge": body.get("challenge")}

    # 3. Callback challenge (không có field type)
    if "challenge" in body and "type" not in body:
        return {"challenge": body.get("challenge")}

    # 4. Parse message event
    message_data = parse_message_event(body)
    if not message_data:
        return {"success": True}

    # 5. Deduplicate
    if is_duplicate(message_data["event_id"]):
        return {"success": True}

    # 6. Xử lý background (tránh webhook timeout)
    asyncio.create_task(process_message(message_data))

    # 7. Trả về ngay
    return {"success": True}

from datetime import datetime, timedelta, timezone

# Timezone Việt Nam (UTC+7)
VN_TIMEZONE = timezone(timedelta(hours=7))

def get_current_time():
    """Lấy thời gian hiện tại theo giờ Việt Nam."""
    return datetime.now(VN_TIMEZONE)

def parse_user_time(time_str: str) -> datetime:
    """Parse input user thành giờ Việt Nam."""
    dt = datetime.strptime(time_str, "%Y-%m-%d %H:%M")
    return dt.replace(tzinfo=VN_TIMEZONE)

async def handle_webhook(request: Request):
    # ... parse event ...

    # Không await -- chạy background
    asyncio.create_task(process_agent_message(
        user_message=message_data["text"],
        session_id=message_data["sender_id"],
        message_id=message_data["message_id"],
        open_id=message_data["open_id"],
    ))

    # Trả về ngay
    return {"success": True}

CREATE TABLE lark_user_tokens (
    id SERIAL PRIMARY KEY,
    open_id VARCHAR(64) UNIQUE NOT NULL,
    user_id VARCHAR(64),
    access_token TEXT NOT NULL,
    refresh_token TEXT NOT NULL,
    expires_at BIGINT NOT NULL,  -- Unix timestamp tính bằng milliseconds
    primary_calendar_id VARCHAR(255),
    scopes TEXT,
    created_at TIMESTAMP DEFAULT NOW(),
    updated_at TIMESTAMP DEFAULT NOW()
);

CREATE INDEX idx_lark_user_tokens_open_id ON lark_user_tokens(open_id);

CREATE TABLE agno_sessions (
    session_id TEXT PRIMARY KEY,
    agent_id TEXT,
    user_id TEXT,
    memory JSONB,
    agent_data JSONB,
    user_data JSONB,
    session_data JSONB,
    created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
    updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
);

CREATE INDEX idx_agno_sessions_user_id ON agno_sessions(user_id);

postgresql+psycopg2://postgres:[password]@[host]:5432/postgres

# Bắt buộc
ANTHROPIC_API_KEY=sk-ant-xxx
LARK_APP_ID=cli_xxx
LARK_APP_SECRET=xxx
LARK_OAUTH_REDIRECT_URI=https://your-domain.com/oauth/callback
SUPABASE_DB_URL=postgresql+psycopg2://...

# Tùy chọn
LARK_ENCRYPT_KEY=xxx
LARK_VERIFICATION_TOKEN=xxx
HOST=0.0.0.0
PORT=7778
AGENT_DEBUG_MODE=false

{
  "build": {
    "builder": "DOCKERFILE",
    "dockerfilePath": "Dockerfile"
  },
  "deploy": {
    "numReplicas": 1,
    "healthcheckPath": "/webhook/health",
    "healthcheckTimeout": 30
  }
}

FROM python:3.13-slim

WORKDIR /app

RUN apt-get update && apt-get install -y gcc curl && rm -rf /var/lib/apt/lists/*

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 7778

CMD ["python", "agent.py"]

# Terminal 1
python agent.py

# Terminal 2
ngrok http 7778

# Sai
GET /calendar/v4/calendars/primary/events

# Đúng
GET /calendar/v4/calendars  # List trước
# Tìm calendar có type="primary", lấy calendar_id thực
GET /calendar/v4/calendars/{actual_calendar_id}/events

from datetime import timezone, timedelta

VN_TZ = timezone(timedelta(hours=7))
now = datetime.now(VN_TZ)

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
import asyncio

app = FastAPI()

@app.post("/webhook/event")
async def webhook(request: Request):
    body = await request.json()

    # Challenge verification
    if body.get("type") == "url_verification":
        return {"challenge": body.get("challenge")}

    if "challenge" in body and "type" not in body:
        return {"challenge": body.get("challenge")}

    # Xử lý tin nhắn background
    asyncio.create_task(process_message(body))

    return {"success": True}

@app.get("/webhook/health")
async def health():
    return {"status": "healthy"}

@app.get("/oauth/callback")
async def oauth_callback(code: str, state: str):
    # Đổi code lấy tokens
    tokens = await exchange_code(code)

    # Lấy thông tin user
    user_info = await get_user_info(tokens["access_token"])
    open_id = user_info.get("open_id") or state

    # Lưu tokens
    save_tokens(open_id, tokens)

    return HTMLResponse("<h1>Đăng nhập thành công!</h1>")