Functions

A function is server TypeScript, RPC-callable from the client. Pylon runs them in a Bun process managed by the runtime. Three flavors:

Query — read-only, can subscribe to changes
Mutation — writes through the transactional path
Action — arbitrary side effects (HTTP calls, emails, file ops)

Writing a function

Create a file in functions/:

// functions/createMessage.ts
import { mutation, v } from "@pylonsync/functions";

export default mutation({
  // Every function declares an auth mode. "user" is the default;
  // see the Auth section below. The framework enforces this before
  // the handler runs, so `ctx.auth.userId` is `string` here.
  args: {
    roomId: v.id("Room"),
    body: v.string(),
  },
  async handler(ctx, args) {
    const id = await ctx.db.insert("Message", {
      roomId: args.roomId,
      authorId: ctx.auth.userId,
      body: args.body,
      sentAt: new Date().toISOString(),
    });
    return { id };
  },
});

The filename becomes the RPC name — createMessage is callable at POST /api/fn/createMessage.

Context object

ctx gives you:

ctx.auth.userId   // string (when auth: "user" | "admin") OR string | null (when auth: "public" | "guest")
ctx.auth.isAdmin  // boolean
ctx.auth.tenantId // string | null  (selected org)
ctx.auth.elevate({ admin: true, reason: "..." })  // promote AFTER verifying a webhook/HMAC
// No ctx.auth.email / ctx.auth.roles on the handler ctx — fetch the user row for email:
//   const u = await ctx.db.get("User", ctx.auth.userId)

// ctx.db — on queries (read-only) and mutations (read + write). NOT on actions.
ctx.db.insert(entity, data)
ctx.db.get(entity, id)
ctx.db.query(entity, filter?)
ctx.db.list(entity)                   // whole entity (no filter)
ctx.db.update(entity, id, patch)
ctx.db.delete(entity, id)
ctx.db.link(entity, id, relation, targetId)
ctx.db.unlink(entity, id, relation)

// Mutations + actions only (not queries):
ctx.error(code, message)              // throw typed error → mutation rolls back
ctx.scheduler.runAfter(delayMs, ...)  // schedule (deferred until COMMIT)
ctx.llm.complete(request)             // LLM completion, returns when the model finishes
ctx.llm.stream(request, onEvent)      // LLM completion, event per delta
ctx.stream.write(text)                // SSE chunk → the one client holding this request
ctx.rooms.broadcast(room, topic, data) // push → every subscriber of a presence room

The three flavors expose different context surfaces:

Query — ctx.db (read-only), ctx.auth, ctx.env, ctx.requireMember.
Mutation — the above plus writes on ctx.db, ctx.scheduler, ctx.error, ctx.llm, ctx.connections, ctx.stream, ctx.rooms.
Action — no ctx.db. Reach the database through ctx.runQuery(name, args) / ctx.runMutation(name, args) (each runs as its own transaction). Actions also get ctx.scheduler, ctx.error, ctx.email, ctx.llm, ctx.stream, ctx.rooms, ctx.connections, and ctx.request (raw HTTP request — the only ctx that has it, so webhook signature checks live in actions).

A mutation handler IS the transaction — every ctx.db.* call inside the handler shares one BEGIN/COMMIT, and a thrown error rolls everything back atomically. There’s no ctx.db.transact([...]) because the handler already wraps your writes; if you need batch atomicity from outside a mutation, use the HTTP /api/transact endpoint or call runMutation from an action.

Auth (secure by default)

Every function declares who can call it. The framework enforces this before the handler runs — a missing if (!ctx.auth.userId) check no longer leaks data, because the runtime made the check first.

export default mutation({
  auth: "user",   // default — signed-in user required
  args: { ... },
  async handler(ctx, args) {
    // ctx.auth.userId is `string`, not `string | null`. The
    // redundant null check is gone.
    await ctx.db.insert("Note", { authorId: ctx.auth.userId, ... });
  },
});

Four modes:

Mode	Reaches the handler when…	Use it for
`"user"` (default)	Real signed-in user. Guest sessions rejected.	Almost everything.
`"public"`	Anyone, including unauthenticated callers.	Healthchecks, landing-form submits, webhook receivers that verify their own signature.
`"guest"`	Guest session OR real user.	Cart-style pre-login state.
`"admin"`	`ctx.auth.isAdmin === true`.	Ops endpoints, dangerous helpers exposed via `/api/fn/...`.

When the mode doesn’t match the caller, the request is rejected before the handler runs — 401 AUTH_REQUIRED for "user" / "guest", 403 FORBIDDEN for "admin". Admin sessions bypass every mode (same convention as policies).

// Public endpoint — has to be explicit, never the default.
export default action({
  auth: "public",
  async handler(ctx) {
    // ctx.auth.userId is `string | null`. Check it if you need it.
    return { ok: true };
  },
});

Action authentication. Policies gate ctx.db.* reads and writes, but they do not gate action handlers. An action that charges Stripe, sends email, or calls a private API relies on its auth setting, so Pylon requires authentication by default. internal: true functions ignore auth — they’re unreachable over HTTP and inherit the wrapping handler’s context.

`ctx.db` honors policies (strict mode)

By default, ctx.db.* inside a function bypasses entity policies because server code is trusted. This remains the default for compatibility. Strict mode flips the default: every ctx.db.get/query/insert/update/delete/lookup/search runs through the policy engine using the function’s caller auth, exactly as if the same operation came in through /api/entities/*. Enable per deploy:

PYLON_STRICT_FN_POLICIES=1 pylon dev

When strict mode is on, the canonical Pylon-shaped IDOR — “I wrote getRecording.ts that does ctx.db.get('Recording', args.id) and forgot to check tenant ownership” — becomes impossible. The policy on Recording fires, sees that the caller isn’t a member of the row’s org, and the call returns POLICY_DENIED before the row leaves the database. For the legitimate cross-tenant cases (admin tools, webhook receivers post signature verification, scheduled cron sweeps), use the explicit escape hatch:

// functions/stripeWebhook.ts
// The webhook is an action — only actions get ctx.request (the raw
// body a signature check needs). Actions have no ctx.db, so the DB
// work runs in an internal mutation the action hands off to.
export default action({
  auth: "public",  // webhook — provider doesn't sign in
  async handler(ctx) {
    const sig = ctx.request!.headers["stripe-signature"];
    const ok = verifyStripeSignature(secret, ctx.request!.rawBody, sig);
    if (!ok) throw ctx.error("INVALID_SIGNATURE", "bad sig");
    // Trust boundary crossed — the webhook is the trusted caller.
    await ctx.runMutation("settleStripeEvent", { customerId });
  },
});

// functions/settleStripeEvent.ts — internal: unreachable over HTTP
export default mutation({
  internal: true,
  args: { customerId: v.string() },
  async handler(ctx, args) {
    // Use ctx.db.unsafe to read any org's row regardless of who's
    // calling. Comment justifies the bypass for code review.
    const org = await ctx.db.unsafe.lookup("Org", "stripeCustomerId", args.customerId);
    // ...
  },
});

Rules of thumb:

Plain ctx.db.* — the default. Acts as the caller. Use for anything that should reflect the user’s view of the data.
ctx.db.unsafe.* — explicit bypass. Use for webhooks, cron sweeps, admin tooling, and anything that genuinely needs cross-tenant reads. Every call should have a justifying comment.
Admin contexts bypass strict mode — auth.isAdmin === true skips the gate the same way it bypasses entity-route policies. Ops scripts and the ctx.auth.elevate({ admin: true, reason: "..." }) path inside verified webhooks still work everywhere (the reason is mandatory and audited).

Strict mode is opt-in for now (one minor cycle) so existing apps can migrate at their own pace. The rollout plan:

Now (v0.3.161+) — ctx.db.unsafe.* is callable. Mark known cross-tenant paths with it. Plain ctx.db.* still bypasses policies (existing behavior); strict mode is opt-in via env.
v0.4 — strict mode default-on. Apps that didn’t migrate get POLICY_DENIED on the calls that genuinely need cross-tenant access; the fix is to mark those unsafe.

Validators

v.* describes expected argument shapes:

v.string()              v.int()                v.float()
v.boolean()             v.datetime()           v.id("User")
v.optional(v.string())  v.array(v.string())    v.literal("open" | "closed")
v.object({ k: v.string() })

The same validators work in query, mutation, and action arguments.

Queries

// functions/listIssues.ts
import { query, v } from "@pylonsync/functions";

export default query({
  args: { teamId: v.id("Team") },
  async handler(ctx, args) {
    return ctx.db.query("Issue", { teamId: args.teamId });
  },
});

Queries are live by default — the React client subscribes and re-runs on relevant changes. See Live queries.

Actions

Use for side effects outside the database:

// functions/sendEmail.ts
import { action, v } from "@pylonsync/functions";

export default action({
  args: { to: v.string(), subject: v.string() },
  async handler(ctx, args) {
    await fetch("https://api.resend.com/emails", {
      method: "POST",
      headers: { Authorization: `Bearer ${process.env.RESEND_KEY}` },
      body: JSON.stringify({ to: args.to, subject: args.subject }),
    });
    return { ok: true };
  },
});

Actions have no ctx.db. They reach the database through ctx.runQuery(name, args) / ctx.runMutation(name, args) — each of those runs as its own transaction, but the action as a whole is not atomic. If a sequence of writes must commit or roll back together, do them inside a single mutation and have the action call it once.

Streaming LLM output

ctx.llm.complete(request) sends a completion and resolves once the model has finished generating. ctx.llm.stream(request, onEvent) sends the same request but calls onEvent for each event as the provider emits it, then resolves with the same assembled response complete would have returned — so a handler can push text to the client as it arrives and still inspect stop_reason afterwards. Both live on mutation + action ctx. Neither is on query ctx: a subscribed query re-runs whenever its ctx.db reads change, and a paid, non-deterministic call has no business re-running on every dep invalidation.

// functions/ask.ts
import { action, v } from "@pylonsync/functions";

export default action({
  args: { question: v.string() },
  async handler(ctx, args) {
    const res = await ctx.llm.stream(
      { messages: [{ role: "user", content: args.question }] },
      (e) => {
        // Push each token to the caller as the model produces it.
        if (e.type === "text_delta") ctx.stream.write(e.text);
      },
    );
    return { stopReason: res.stop_reason, usage: res.usage };
  },
});

The client reads the chunks with db.streamFn:

for await (const chunk of db.streamFn("ask", { question: "why is the sky blue?" })) {
  setAnswer((prev) => prev + chunk);
}

Stream events

onEvent receives one of four shapes:

Event	Fields	What to do with it
`text_delta`	`text`	Append it — this is the model’s prose, one fragment at a time.
`tool_use_start`	`id`, `name`	A tool call is opening. Start a buffer for its arguments.
`tool_input_delta`	`partial_json`	A raw JSON fragment of the open tool’s arguments. Concatenate; parse once, at the end. Individual fragments are not valid JSON.
`done`	`stop_reason`, `usage`	Always fires last, including on a partial failure.

Everything else matches complete: same auth gating, same model allowlist (PYLON_AI_MODELS_ALLOWED / the manifest llm() helper), same thrown errors carrying err.code (LLM_NOT_CONFIGURED, MODEL_NOT_ALLOWED, PROVIDER_HTTP_429, …). Streaming can’t reach a model complete would refuse.

Streaming does not extend the deadline

The call deadline is absolute wall clock from invocation — PYLON_FN_CALL_TIMEOUT, 30 seconds by default. Emitting events does not reset it. A multi-turn agent that runs tools will blow through 30s, so declare a timeout on the function:

export default action({
  timeout: 300,   // seconds of wall clock for the whole handler
  args: { ... },
  async handler(ctx, args) { ... },
});

Agent tool loop

Stream the text out as it’s generated, run tools when the model asks, feed the results back, repeat until the model stops asking:

// functions/agent.ts
import { action, v } from "@pylonsync/functions";

// Anthropic Messages shape: content is prose, or content blocks —
// tool_use coming back from the model, tool_result going the other way.
type Turn = { role: "user" | "assistant"; content: string | any[] };

const tools = [
  {
    name: "get_order",
    description: "Look up an order by id.",
    input_schema: {
      type: "object",
      properties: { orderId: { type: "string" } },
      required: ["orderId"],
    },
  },
];

export default action({
  args: { question: v.string() },
  timeout: 300,   // streaming does NOT extend the 30s default
  async handler(ctx, args) {
    const messages: Turn[] = [{ role: "user", content: args.question }];

    for (let turn = 0; turn < 10; turn++) {
      const res = await ctx.llm.stream({ messages, tools }, (e) => {
        if (e.type === "text_delta") ctx.stream.write(e.text);
      });

      // Keep the assistant turn verbatim — it carries the tool_use
      // blocks the next turn's tool_result blocks refer to by id.
      messages.push({ role: "assistant", content: res.content });

      if (res.stop_reason !== "tool_use") return { turns: turn + 1 };

      const results = [];
      for (const block of res.content) {
        if (block.type !== "tool_use") continue;
        const order = await ctx.runQuery("getOrder", {
          id: (block.input as { orderId: string }).orderId,
        });
        results.push({
          type: "tool_result" as const,
          tool_use_id: block.id,
          content: JSON.stringify(order ?? { error: "not found" }),
        });
      }
      messages.push({ role: "user", content: results });
    }

    throw ctx.error("AGENT_LOOP_LIMIT", "tool loop did not converge");
  },
});

The loop bound matters: a model can keep requesting tools indefinitely, and without a cap the only thing that stops it is the call deadline killing the worker mid-turn.

Server push to a room

ctx.stream.write writes to the HTTP response of the call in flight. It reaches exactly one client — the one that made the request. Close the tab and the output is gone; a second device never sees it. ctx.rooms.broadcast(room, topic, data) pushes to every subscriber of a presence room over the WebSocket, independent of who made the call. Same rooms clients join with useRoom, same delivery path a member’s own broadcast() uses.

await ctx.rooms.broadcast("chat-42", "agent.delta", { text: "hi" });
// => { delivered: true }

It resolves { delivered: false } when the room has no members. Broadcasting into an empty room is a no-op, not an error — an agent doesn’t have to know whether anyone is watching. Available on mutation + action ctx, not on queries: a reactive query re-runs on every dep change, which would re-broadcast each time. Pick based on who needs the output:

One caller, one screen (a chat box waiting on its own response) — ctx.stream.write, read with db.streamFn.
Everyone watching, including tabs that weren’t the caller (agent output that must survive a reload, a shared session on a second device, a job with no HTTP caller at all) — ctx.rooms.broadcast.

Doing both in one handler is normal — write to the caller for the low-latency path, broadcast so every other watcher stays in sync:

await ctx.llm.stream({ messages }, (e) => {
  if (e.type !== "text_delta") return;
  ctx.stream.write(e.text);                                  // the caller
  void ctx.rooms.broadcast(room, "agent.delta", { text: e.text }); // everyone else
});

On the client, join the room with useRoom and read the pushed messages off the sync engine:

import { useRoom, getSync } from "@pylonsync/react";

useRoom("chat-42", userId);   // joins the room — required to receive pushes

useEffect(() => {
  return getSync().subscribeRoomMessages("chat-42", (m) => {
    if (m.topic === "agent.delta") append((m.payload as { text: string }).text);
  });
}, []);

Server broadcasts arrive with an empty from (no sender user id) — a member’s own broadcasts carry theirs, so m.from === userId is how you filter your own echoes.

Calling functions from the client

import { callFn } from "@pylonsync/react";

async function onSend() {
  const { id } = await callFn("createMessage", {
    roomId: "01H...",
    body: "hello",
  });
  console.log("created", id);
}

Errors

Throw typed errors that propagate to the client with structured codes:

if (!ctx.auth.userId) {
  throw ctx.error("UNAUTHENTICATED", "log in first");
}
if (msg.length > 2000) {
  throw ctx.error("INVALID_ARGS", "message too long");
}

The client receives { code, message } and can render different UI for each code.

Live queries

How query subscriptions stay in sync.

Validators

All argument shapes v.* supports.

Get started

Core concepts

SSR

Auth

Plugins

Clients

Operations

Compare

Writing a function

Context object

Auth (secure by default)

`ctx.db` honors policies (strict mode)

Validators

Queries

Actions

Streaming LLM output

Stream events

Streaming does not extend the deadline

Agent tool loop

Server push to a room

Calling functions from the client

Errors

Next

Live queries

Validators

​Writing a function

​Context object

​Auth (secure by default)

​ctx.db honors policies (strict mode)

​Validators

​Queries

​Actions

​Streaming LLM output

​Stream events

​Streaming does not extend the deadline

​Agent tool loop

​Server push to a room

​Calling functions from the client

​Errors

​Next

Live queries

Validators

Writing a function

Context object

Auth (secure by default)

`ctx.db` honors policies (strict mode)

Validators

Queries

Actions

Streaming LLM output

Stream events

Streaming does not extend the deadline

Agent tool loop

Server push to a room

Calling functions from the client

Errors

Next