Python: an AI agent that writes to each user's Notion, with FastAPI

To let a Python AI agent write to each user's own Notion, have each user connect Notion once, then make the agent's calls with client.for_user(user_id). Anlyon holds the user's Notion token, makes the call, counts it against that user's daily limit and reads the page back to grade the receipt.

A user says "save these meeting notes to my Notion". Your FastAPI service runs the agent, the agent calls a tool, and a page appears in that user's workspace, under the parent page they picked. This tutorial covers the routes, the tool, the limit, the receipts and account deletion. For the model behind it, read the pillar guide on letting your AI agent act in your users' accounts.

What you will build

• A FastAPI route that sends a user to connect their own Notion, and a callback that confirms it.
• An agent tool that creates a Notion page with a paragraph of content in that user's workspace.
• A per-user limit: 20 new Notion pages a day for each user.
• A route that lists one user's receipts.
• An account-deletion hook that erases the user from Anlyon.

Prerequisites

• Python 3.10 or later, FastAPI, and a way to know who is signed in. The code uses a current_user dependency; use whatever your auth gives you.
• An Anlyon workspace with an operator key (actions:write, impact-limits:write) and an app key from the Standard preset plus end-users:write, with Require an end user on.
• A Notion workspace with one page you can share with the integration.

pip install fastapi uvicorn anlyon

1. Install the Notion pack

The Python SDK has no pack installer, so use the CLI once, with the operator key:

ANLYON_API_KEY=$ANLYON_OPERATOR_KEY npx @anlyonhq/cli packs install notion --yes

This adds Notion's reads and writes as actions with the notion- prefix, including notion-search-pages, notion-create-page, notion-append-paragraph and notion-update-page-properties. Each write declares one resource_mutations and a read-back. Each request sends Authorization: Bearer {{connection:notion}}, which resolves to the calling user's own token. See the packs docs.

2. Set a per-user limit

# setup_anlyon.py, run once
import os
from anlyon import Client

ops = Client(api_key=os.environ["ANLYON_OPERATOR_KEY"])

ops.impact_limits.create(
    name="Notion pages per user per day",
    dimension="resource_mutations",
    period="day",
    amount=20,
    scope="end_user",
    action_name="notion-create-page",
)

scope="end_user" gives each user their own 20 a day, counted per UTC day and reserved before the call. The 21st page for one user is refused before anything reaches Notion, and every other user keeps their allowance. Creating a limit tightens a guardrail, so it applies at once. More in per-user limits for AI agents.

3. The connect route and the callback

# app.py
import os

from fastapi import Depends, FastAPI, Request
from fastapi.responses import RedirectResponse
from anlyon import Client, EndUserMismatchError, parse_connect_redirect

from .auth import User, current_user  # your own session

anlyon = Client(api_key=os.environ["ANLYON_API_KEY"])
app = FastAPI()

CALLBACK = "https://app.example.com/connections/callback"


@app.get("/connect/notion")
def connect_notion(user: User = Depends(current_user)):
    link = anlyon.end_users.create_connect_link(
        end_user_id=user.id, integration="notion", redirect_url=CALLBACK
    )
    return RedirectResponse(link.data["url"])  # works once, for 10 minutes


@app.get("/connections/callback")
def connections_callback(request: Request, user: User = Depends(current_user)):
    result = parse_connect_redirect(str(request.url))
    if not result or result.status != "pending_confirmation" or not result.flow_id:
        return RedirectResponse("/settings?notion=failed")  # declined or error
    try:
        # The user from your session, never one read from the URL.
        anlyon.end_users.confirm_connection(user.id, result.flow_id)
    except EndUserMismatchError:
        return RedirectResponse("/settings?notion=wrong-user")
    return RedirectResponse("/settings?notion=connected")

The user sees Anlyon's consent page, which names your workspace, then Notion's, where they pick the pages the integration may reach. Notion has no OAuth scopes: the pages they pick are what the agent can write under. The connection exists once your server confirms it. A link forwarded to someone else cannot attach their Notion to your user; the confirm raises EndUserMismatchError and connects nothing.

Users who prefer an internal integration token can paste it on Anlyon's hosted form instead, or your server can store one it already has with anlyon.end_users.set_connection(user.id, "notion", token=token).

4. The agent's tool

The tool takes what the model chooses: a parent page, a title and the text. It does not take a user. Your server binds the user before the agent runs.

# notion_tool.py
import json
from typing import Any

from anlyon import Client, EndUserConnectionRequiredError

SAVE_TO_NOTION = {
    "name": "save_to_notion",
    "description": "Create a page in the user's Notion under a parent page, with one paragraph of text.",
    "input_schema": {
        "type": "object",
        "required": ["parent_page_id", "title", "text"],
        "properties": {
            "parent_page_id": {"type": "string", "description": "A Notion page id the user shared."},
            "title": {"type": "string", "maxLength": 200},
            "text": {"type": "string", "maxLength": 2000},
        },
    },
}


def save_to_notion(anlyon: Client, user_id: str, parent_page_id: str, title: str, text: str) -> dict[str, Any]:
    user = anlyon.for_user(user_id)  # fixed to this user; it cannot be switched
    try:
        page = user.actions.invoke(
            "notion-create-page", {"parent_page_id": parent_page_id, "title": title}
        ).data
    except EndUserConnectionRequiredError as error:
        # Not connected, or revoked: nothing was sent.
        return {"status": error.code.lower(), "connect_url": error.connect_url}

    if page["status"] != "succeeded":
        return {"status": page["status"], "grade": page.get("grade"), "error": page.get("error")}

    page_id = json.loads(page["body"])["id"]
    paragraph = user.actions.invoke("notion-append-paragraph", {"page_id": page_id, "text": text}).data
    return {
        "status": paragraph["status"],
        "page_id": page_id,
        "grades": [page.get("grade"), paragraph.get("grade")],
    }

The pack's create-page takes parent_page_id (a Notion id, 32 to 36 hex characters and hyphens) and title (one line, up to 200 characters). append-paragraph takes page_id and text up to 2,000 characters. Anything else is refused before the call.

Wire the tool into the agent from a route that knows who is signed in. Any tool-calling loop works: hand the model SAVE_TO_NOTION, and when it calls the tool, run it with the user from the session:

@app.post("/agent/tools/save_to_notion")
def run_save_to_notion(args: dict, user: User = Depends(current_user)):
    return save_to_notion(anlyon, user.id, **args)

The user id comes from current_user, never from args. If your code passed a user id taken from model output into for_user, a prompt could pick the account, and the binding would no longer protect you.

5. List a user's receipts

@app.get("/me/notion-receipts")
def my_receipts(user: User = Depends(current_user)):
    page = anlyon.actions.effects(end_user_id=user.id, action="notion-create-page", limit=50)
    return [
        {"id": r["id"], "at": r["createdAt"], "outcome": r["outcome"], "grade": r["grade"]}
        for r in page.data
    ]

Each receipt is graded: confirmed when Anlyon read the new page back from Notion and its title matched, acknowledged when Notion accepted it and no read-back matched, unknown when no usable response came back, and failed, refused or denied when it did not run. Filter with grade="unknown" to find pages to check. The guide to querying one user's agent actions covers invocations and pagination.

6. Erase the user when they delete their account

@app.delete("/me")
def delete_account(user: User = Depends(current_user)):
    erased = anlyon.end_users.erase(user.id).data
    # erased: {"connectionsDeleted", "approvalsCancelled", "countersCleared", "erasedAs"}
    delete_user_from_your_db(user.id)
    return {"erased": True, "connections_deleted": erased["connectionsDeleted"]}

erase revokes the user's connections at the provider where it supports that, then deletes them, cancels their pending approvals and clears their limit counters. Invocations, receipts, approvals and audit records are kept, with a keyed one-way hash (erasedAs) in place of the user id. That is what it deletes and what it keeps; how it fits your own deletion policy is your call.

Test it

1. Before connecting, call the tool route with a parent page id. It returns {"status": "end_user_not_connected", "connect_url": "https://..."}. Nothing was sent to Notion.
2. Visit /connect/notion, pick a test page on Notion's consent screen, and land on /settings?notion=connected. anlyon.end_users.list_connections(user.id) shows notion with status active.
3. Call the tool again with that page id. It returns status succeeded and grades ["confirmed", "confirmed"]: Anlyon read the page back and its title matched, then read the paragraph back and its text matched. The page is in the user's workspace.
4. Call it 20 more times today as the same user. The 21st notion-create-page comes back with status failed, the error Dispatch refused (impact_limit_exceeded), and a receipt graded refused. A second user still creates pages.
5. GET /me/notion-receipts lists each page with its grade.
6. DELETE /me, then call the tool again: it returns end_user_not_connected, and the receipts in the console show a hash where the user id was.

Limits

• The agent reaches the pages the user shared with the integration on Notion's consent screen, and no others. A parent page they did not share fails at Notion.
• Notion has no idempotency key for these writes. Anlyon never resends a call that ended unknown, so check Notion before asking the agent again.
• The limit counts calls made through Anlyon. A Notion call your code makes with its own token has no limit and no receipt.
• erase revokes at the provider where the provider supports it. It does not delete pages already written to the user's Notion.

Next steps

• Act for your users: the full model, including approvals decided by the user.
• End-user approval in your app: add decided_by="end_user" to a policy for a step like archiving a page.
• Vercel AI SDK tools that act in each user's account, the TypeScript version.
• Docs: Act for your users · Packs