Author a connector

Connectors are how Lobu turns external systems (REST APIs, GraphQL, webhooks, files, OAuth-protected services) into the typed event stream that behaviors shape into entities and memory.

A connector is a TypeScript class that extends ConnectorRuntime and ships three things:

a definition describing the connector (key, name, version, auth, feeds, actions),
a sync(ctx) method that pulls the next slice of data and returns events,
an optional execute(ctx) method that runs writes back to the source (create issue, send email).

Sync runs are idempotent: each run returns a checkpoint (cursor, timestamp, ID set) that the next run reads back via ctx.checkpoint.

Both authoring surfaces live in the same @lobu/connector-sdk package. This page covers connectors first, then reactions — the governed TypeScript hooks that run after a behavior extracts data. The type reference for everything below is the Connector SDK reference.

Install

bun add @lobu/connector-sdk
# or
npm install @lobu/connector-sdk
# or
pnpm add @lobu/connector-sdk

A typed connector, end to end

The example below pulls issues from a GitHub repository, polls incrementally with a typed checkpoint, and emits one EventEnvelope per issue. Every field has a real type: no as any casts, no // biome-ignore directives.

import {
  ConnectorRuntime,
  Type,
  type Static,
  type ConnectorDefinition,
  type EventEnvelope,
  type SyncContext,
  type SyncResult,
} from "@lobu/connector-sdk";

// User-supplied feed config, rendered as a form in the admin UI.
const GitHubConfigSchema = Type.Object({
  owner: Type.String({ description: "GitHub organization or user" }),
  repo: Type.String({ description: "Repository name" }),
});

type GitHubConfig = Static<typeof GitHubConfigSchema> & {
  // Added by the env_keys auth method at execution time.
  token?: string;
};

// The timestamp watermark persisted between runs, so each sync asks only for
// issues updated since the last successful one.
interface GitHubCheckpoint {
  last_updated_at: string | null;
}

// Minimal subset of the GitHub REST API issue payload we actually read.
interface GitHubIssue {
  id: number;
  number: number;
  title: string;
  body: string | null;
  html_url: string;
  updated_at: string;
  user: { login: string } | null;
}

export default class GitHubIssuesConnector extends ConnectorRuntime<
  GitHubCheckpoint,
  GitHubConfig
> {
  readonly definition: ConnectorDefinition = {
    key: "github-issues",
    name: "GitHub issues",
    version: "1.0.0",
    // Personal access token collected for each connection.
    authSchema: {
      methods: [
        {
          type: "env_keys",
          fields: [
            { key: "token", label: "GitHub PAT", secret: true, required: true },
          ],
        },
      ],
    },
    feeds: {
      issues: {
        key: "issues",
        name: "Issues",
        configSchema: GitHubConfigSchema,
      },
    },
  };

  async sync(
    ctx: SyncContext<GitHubCheckpoint, GitHubConfig>
  ): Promise<SyncResult<GitHubCheckpoint>> {
    // For `env_keys` auth, the values land in `ctx.config` keyed by the
    // `key` you declared on the auth field. OAuth tokens (for `oauth` auth)
    // arrive on `ctx.credentials.accessToken` instead.
    const config = ctx.config;
    const checkpoint = ctx.checkpoint ?? { last_updated_at: null };
    const token = config.token ?? "";

    // GitHub returns issues updated *at or after* `since`, so discard the
    // checkpoint boundary after fetching.
    const since = checkpoint.last_updated_at ?? "1970-01-01T00:00:00Z";
    const url =
      `https://api.github.com/repos/${config.owner}/${config.repo}/issues` +
      `?state=all&sort=updated&direction=asc&per_page=100&since=${since}`;

    const response = await fetch(url, {
      headers: {
        Authorization: `Bearer ${token}`,
        Accept: "application/vnd.github+json",
      },
    });
    if (!response.ok) {
      throw new Error(`GitHub ${response.status}: ${await response.text()}`);
    }

    const issues = (await response.json()) as GitHubIssue[];
    const fresh = issues.filter((i) => i.updated_at !== checkpoint.last_updated_at);

    const events: EventEnvelope[] = fresh.map((issue) => ({
      origin_id: String(issue.id),
      origin_type: "issue",
      title: `#${issue.number} ${issue.title}`,
      payload_text: issue.body ?? "",
      source_url: issue.html_url,
      author_name: issue.user?.login,
      occurred_at: new Date(issue.updated_at),
    }));

    return {
      events,
      // Always advance the checkpoint to the newest `updated_at` we saw.
      // If the page was empty, return the previous value verbatim so the
      // next run is still idempotent.
      checkpoint: {
        last_updated_at:
          fresh.at(-1)?.updated_at ?? checkpoint.last_updated_at,
      } satisfies GitHubCheckpoint,
    };
  }
}

A few things to notice:

The ConnectorRuntime generics carry your types through the whole sync. ctx.checkpoint is GitHubCheckpoint | null, and ctx.config is GitHubConfig.
env_keys credentials live on ctx.config, not ctx.credentials. Lobu merges the values the user filled into the env_keys form into ctx.config under the keys you declared (token here). ctx.credentials is reserved for oauth auth: accessToken, refreshToken, scope, expiresAt.
Secret values stay behind Lobu’s credential proxy. Connector code uses the value from ctx.config, while Lobu resolves the credential only for the allowed outbound request. See Secret proxy.
Incremental reads via the since query param. Follow GitHub’s Link header as well when one sync can return more than a page; the example stays focused on the checkpoint contract.

Save this file in your Lobu project (e.g. github-issues.connector.ts next to lobu.config.ts) and list it in your config:

import { connectorFromFile, defineConfig } from "@lobu/cli/config";
import type GitHubIssuesConnector from "./github-issues.connector.ts";

export default defineConfig({
  connectors: [
    connectorFromFile<typeof GitHubIssuesConnector>(
      "./github-issues.connector.ts"
    ),
  ],
  // ...agents, connections, etc.
});

Passing the connector’s type via the generic (import type + connectorFromFile<typeof GitHubIssuesConnector>) is optional (bare connectorFromFile("./github-issues.connector.ts") still works), but it gives you go-to-definition, rename, and a tsc error if the file’s default export ever stops being a ConnectorRuntime subclass. The import type is erased at compile time, so the connector module is never loaded while your config is evaluated.

lobu apply ships the source to the gateway, which compiles and registers it; from there each feeds.<key> entry shows up as something a user can create a connection for in the admin UI.

Concepts

`ConnectorDefinition`

The static metadata Lobu publishes for your connector after lobu apply.

Field	Required	Description
`key`	yes	Unique global key, e.g. `google.gmail`, `github-issues`
`name`	yes	Human-readable label
`version`	yes	Semver; bump to invalidate per-feed checkpoints if the event shape changes
`kind`	no	`data` by default; use `integration` for an app/webhook surface with no syncable feeds
`authSchema`	no	How users authenticate this connector (see below)
`feeds`	no	Map of feed key → `FeedDefinition` (a connector typically has one or more feeds)
`actions`	no	Map of action key → `ActionDefinition` (only needed if you also implement `execute`)
`behaviorEvents`	no	Normalized event triggers a behavior can subscribe to
`webhook`	no	Gateway-side signature verification and delivery routing for inbound webhooks
`requiredCapability`	no	When set, only worker pods/devices advertising this capability serve runs (e.g. `screentime` for the Mac app)
`runtime`	no	Pin to a device platform (iOS, macOS, …); omit for cloud-side connectors
`agentTooling`	no	Native packages, leased credentials, and domains the connection contributes to agent sandboxes

See the full type at reference/connector-sdk › ConnectorDefinition.

`SyncContext`

What sync() receives. Every field is read-only.

Field	Description
`feedKey`	Which feed Lobu is asking you to run
`feedId`	Stable ID of this configured feed, when the platform supplies one
`config`	The connection-level config the user filled in (typed by your `FeedDefinition.configSchema`)
`checkpoint`	The last successful run’s checkpoint, or `null` on the first run
`credentials`	OAuth tokens (`accessToken`, `refreshToken`, …) for `oauth` auth; `null` for everything else. `env_keys` values land on `ctx.config` under the declared `key`.
`entityIds`	Entities this feed is linked to (rarely needed; useful for scoping the sync)
`sessionState`	Browser cookies / tokens captured by `lobu memory browser-auth` for `browser` auth
`installation`	App-installation context for connections authenticated through an installed app
`emitEvents(events)`	Optional streaming hook: flush a chunk before the run ends
`updateCheckpoint(cp)`	Optional progress-checkpoint hook for long-running syncs

SyncContext<C, F> is generic: C types the checkpoint and F types the feed config. Pass both types to ConnectorRuntime<C, F> so ctx.checkpoint and ctx.config stay typed without casts.

`EventEnvelope`

The shape of one durable event produced by a connector.

interface EventEnvelope {
  origin_id: string; // platform's unique ID for this item
  origin_type?: string; // source-native type (post, message, charge)
  payload_text: string; // main content
  payload_type?: "text" | "markdown" | "json_template" | "media" | "empty";
  title?: string;
  author_name?: string;
  source_url?: string; // permalink back to the original
  occurred_at: Date; // when the event actually happened
  semantic_type?: string; // content, note, summary, fact, etc.
  score?: number; // 0-100 engagement / relevance
  metadata?: Record<string, unknown>;
}

Only origin_id, payload_text, and occurred_at are required. The full surface is documented in reference/connector-sdk › EventEnvelope.

`SyncResult`

interface SyncResult {
  events: EventEnvelope[];
  checkpoint: Record<string, unknown> | null;
  auth_update?: Record<string, unknown> | null;
  metadata?: {
    items_found?: number;
    items_skipped?: number;
    [key: string]: unknown;
  };
}

Return events: [] plus the same checkpoint you received on a no-new-data tick; runs stay idempotent.

`ActionContext` / `ActionResult`

If your connector also writes back (e.g. assign_issue, send_email), declare an actions map on the definition and implement execute(ctx):

import type { ActionContext, ActionResult } from "@lobu/connector-sdk";

interface AssignIssueInput {
  issueId: string;
  assignee: string;
}

async execute(ctx: ActionContext): Promise<ActionResult> {
  if (ctx.actionKey !== "assign_issue") {
    return { success: false, error: `unknown action ${ctx.actionKey}` };
  }
  const { issueId, assignee } = ctx.input as unknown as AssignIssueInput;
  // Same `env_keys` field as sync(); execute()'s ctx.config carries it too.
  const token = String((ctx.config as { token?: string }).token ?? "");

  await fetch(`https://api.example.com/issues/${issueId}`, {
    method: "PATCH",
    headers: { Authorization: `Bearer ${token}` },
    body: JSON.stringify({ assignee }),
  });
  return { success: true, output: { issueId, assignee } };
}

Each ActionDefinition declares requiresApproval: true | false, an optional kind (read or write), and MCP-style annotations such as readOnlyHint, destructiveHint, openWorldHint, and idempotentHint. Lobu routes approval-gated actions through the approval queue before execution.

Auth models

Declare on definition.authSchema. A connector can list multiple methods; the gateway lets the user pick.

`type`	Use when
`none`	Public endpoint, no credentials needed
`env_keys`	Static API keys (Stripe secret key, PAT); fields rendered as form inputs, stored encrypted
`oauth`	Standard OAuth 2.0; Lobu handles the dance, refresh, and per-user token isolation
`app_installation`	Organization- or workspace-scoped app install with gateway-managed credentials and webhooks
`browser`	Session cookies captured via `lobu memory browser-auth` from a logged-in Chrome profile (or CDP)
`interactive`	Custom auth flow (QR pairing, OTP, signed device handshake); implement `authenticate(ctx)` and stream `AuthArtifact`s

Use credentials from ctx.config or ctx.credentials like normal request values. Lobu keeps the underlying secret behind its credential proxy and resolves it only for allowed outbound requests.

Full breakdown at reference/connector-sdk › ConnectorAuthSchema.

Checkpoints

The checkpoint is your bookmark. Lobu saves it after a successful sync and passes it back as ctx.checkpoint on the next run. Three common shapes:

// Timestamp cursor (GitHub `since`, Stripe `created[gt]`):
interface TimestampCheckpoint {
  last_updated_at: string | null;
}

// Page token (Google APIs):
interface PageTokenCheckpoint {
  next_page_token: string | null;
}

// Bounded ID set (idempotency, no native cursor):
interface IdSetCheckpoint {
  seen_ids: string[];
}

Rules of thumb:

Always return a checkpoint, even on the no-new-data case: return the previous one verbatim. Returning null tells the gateway to treat the next run as a fresh start.
Cap unbounded structures (ID sets, in-flight queues) before persisting. Keep the last 1000 IDs, enough to dedupe across a sync window without bloating the row.
Long-running syncs can call ctx.updateCheckpoint(...) mid-flight so a crash doesn’t lose progress.

Where the file lives

A *.connector.ts file can live anywhere in your Lobu project; reference each one explicitly with connectorFromFile in defineConfig({ connectors }):

my-agent/
├── lobu.config.ts          # connectors: [connectorFromFile<typeof GitHubIssuesConnector>(...)]
├── github-issues.connector.ts
├── stripe-charges.connector.ts
└── agents/my-agent/...

lobu apply type-checks and ships the connectors listed in defineConfig({ connectors }). Update the version field whenever the event shape changes so the gateway forces a fresh checkpoint.

Unlike reactions and MCP run_sdk scripts, a connector does not run in an isolate. It is bundled into the worker at compile time and executes in the worker’s runtime with its declared npm/nix dependencies on PATH — the same trust boundary as the agent it belongs to.

Dependencies

A connector can pull in two kinds of dependency, and they are provisioned differently.

npm packages are bundled at compile time. Add them to the package.json next to your lobu.config.ts and import them normally:

import { parse } from "csv-parse/sync";

Lobu packages those imports with the connector when you run lobu apply.

Native tools are provisioned at run time via nix. Declare them as nixpkgs attribute refs in runtime.nix.packages on the connector definition:

export default class VideoConnector extends ConnectorRuntime {
  definition: ConnectorDefinition = {
    key: "media.video",
    name: "Video",
    version: "1.0.0",
    runtime: {
      platforms: ["linux", "macos"],
      nix: { packages: ["ffmpeg", "imagemagick"] },
    },
    // ...feeds, actions
  };
  // ...sync / execute can now shell out to ffmpeg
}

The selected runtime makes the declared tools available on PATH. Choose a runtime that supports native dependencies.

The rule of thumb: npm is bundled (compile-time), native is nix (run-time). Never put a native tool in package.json expecting it to ship, and never list an npm package in runtime.nix.packages. See the ConnectorRuntimeInfo reference for the field shape.

See it in production

examples/ecommerce/stripe-charges.connector.ts: REST API, env_keys auth, timestamp checkpoint.
examples/lobu-crm/npm-downloads.connector.ts: small public HTTP API, ID-set dedupe.

Reactions

A reaction is TypeScript that runs after a behavior completes a window. Use it when validated behavior output should cause a deterministic action: save a derived event, update an entity, notify people, or call a connector operation.

The behavior supplies judgment. The reaction supplies the typed, repeatable effect.

new events → behavior prompt and skills → validated output → reaction → durable result

Reactions are optional. Without one, Lobu still records the behavior run and promotes configured entity output. Add a reaction only when code needs to do something else.

Install

Reaction types are exported by @lobu/connector-sdk:

bun add @lobu/connector-sdk

import type {
  ReactionClient,
  ReactionContext,
} from "@lobu/connector-sdk";

The runtime provides the live client when the reaction runs.

A complete reaction

This reaction receives a validated incident assessment, saves a durable incident event, and notifies administrators when the incident is critical.

import type {
  ReactionClient,
  ReactionContext,
} from "@lobu/connector-sdk";

// This plain JSON Schema becomes the behavior's output contract.
export const input = {
  type: "object",
  properties: {
    severity: {
      enum: ["low", "medium", "high", "critical"],
    },
    summary: { type: "string" },
    evidence_event_ids: {
      type: "array",
      items: { type: "number" },
    },
  },
  required: ["severity", "summary"],
  additionalProperties: false,
} as const;

interface IncidentAssessment {
  severity: "low" | "medium" | "high" | "critical";
  summary: string;
  evidence_event_ids?: number[];
}

export default async function react(
  ctx: ReactionContext,
  client: ReactionClient
): Promise<void> {
  const assessment = ctx.extracted_data as IncidentAssessment;

  const saved = await client.knowledge.save({
    entity_ids: ctx.entities.map((entity) => entity.id),
    content: assessment.summary,
    semantic_type: "incident",
    idempotency_key: `incident:${ctx.behavior.id}:${ctx.window.id}`,
    behavior_source: {
      behavior_id: ctx.behavior.id,
      window_id: ctx.window.id,
    },
    metadata: {
      severity: assessment.severity,
      evidence_event_ids: assessment.evidence_event_ids ?? [],
      window_id: ctx.window.id,
    },
  });

  if (assessment.severity === "critical") {
    await client.notifications.send({
      title: `Critical incident: ${ctx.behavior.name}`,
      body: assessment.summary,
      recipients: "admins",
      resource_url: `/${ctx.organization_slug}/entities`,
      idempotency_key: `incident-notification:${ctx.behavior.id}:${ctx.window.id}`,
      behavior_source: {
        behavior_id: ctx.behavior.id,
        window_id: ctx.window.id,
      },
    });
  }

  client.log(saved.created ? "Incident saved" : "Incident already recorded", {
    event_id: saved.id,
  });
}

Three details matter:

Export input as plain JSON Schema. Lobu uses it as the extraction contract and validates ctx.extracted_data before the reaction runs.
Import both public types. ReactionContext describes the completed behavior window; ReactionClient describes every SDK method available to the script.
Use stable idempotency keys. Reactions may be retried. knowledge.save and notifications.send collapse retries that reuse the same key.
Throw on failure. A resolved handler is success. A thrown error fails the reaction and appears on the behavior run.

Attach it to a behavior

Reference the file explicitly from lobu.config.ts:

import {
  defineBehavior,
  reactionFromFile,
} from "@lobu/cli/config";
import type incidentReaction from "./reactions/incident.reaction.ts";

const incidentMonitor = defineBehavior({
  agent: operations,
  slug: "incident-monitor",
  name: "Incident monitor",
  triggers: [{ kind: "schedule", cron: "*/15 * * * *" }],
  prompt: "Assess new operational signals and cite the evidence.",
  reaction: reactionFromFile<typeof incidentReaction>(
    "./reactions/incident.reaction.ts"
  ),
});

The typed form checks that the file has a compatible default export. The path is relative to lobu.config.ts and must stay inside the project.

What the client can do

ReactionClient is the supported action surface inside a reaction.

Namespace	Use it for
`client.knowledge`	Search, read, save, or tombstone durable memory events.
`client.entities`	List, create, update, delete, search, and link typed entities.
`client.notifications`	Send an inbox notification and optionally deliver it through active bot connections.
`client.operations`	Discover and execute actions exposed by installed connectors or MCP integrations.
`client.query(sql)`	Run a read-only SQL query against the organization.
`client.log(message, data?)`	Add structured information to the behavior run log.

Save knowledge

const saved = await client.knowledge.save({
  entity_ids: ctx.entities.map((entity) => entity.id),
  content: "Renewal risk moved from medium to high.",
  semantic_type: "health_change",
  idempotency_key: `health-change:${ctx.behavior.id}:${ctx.window.id}`,
  behavior_source: {
    behavior_id: ctx.behavior.id,
    window_id: ctx.window.id,
  },
  metadata: { window_id: ctx.window.id },
});

// { id, created, metadata }
client.log(saved.created ? "Health change saved" : "Retry deduplicated");

Update an entity

await client.entities.update({
  entity_id: ctx.entities[0]!.id,
  metadata: { risk: "high", reviewed_at: new Date().toISOString() },
});

Execute a connector operation

const result = await client.operations.execute({
  connection_id: 42,
  operation_key: "create_issue",
  input: {
    title: "Investigate critical incident",
    body: String(ctx.extracted_data.summary ?? ""),
  },
});

if (result.status === "pending_approval") {
  client.log("Issue creation is waiting for approval", {
    run_id: result.run_id,
  });
}

Connector operations still follow the owning agent’s tool policy. A gated operation returns pending_approval; treat that as a valid queued outcome rather than a failure.

Reaction context

interface ReactionContext {
  extracted_data: Record<string, unknown>;
  entities: Array<{
    id: number;
    name: string;
    entity_type: string;
    metadata: Record<string, unknown>;
  }>;
  window: {
    id: number;
    run_id?: number | null;
    behavior_id: number;
    window_start: string;
    window_end: string;
    granularity: string;
    content_analyzed: number;
  };
  behavior: {
    id: number;
    slug: string;
    name: string;
    version: number;
  };
  organization_id: string;
  organization_slug: string;
}

Use ctx.extracted_data for the validated behavior output, ctx.entities for the entities bound to the window, and ctx.window for provenance. ctx.organization_slug is the stable workspace slug for relative links back into Lobu.

Runtime behavior

Reactions compile to JavaScript with esbuild and run in an isolated V8 isolate (isolated-vm), not in the connector or worker process. The same runScript runner backs MCP query_sdk/run_sdk (MCP reference); reactions get a 60 s budget. SDK calls are bridged to the host, so the script has no direct network, filesystem, or process access.
SDK calls are attributed to the behavior and evaluated against its owning agent’s autonomous-action policy.
Network requests follow the agent’s network policy. Prefer client.operations.execute when an installed connector already owns the integration and its credentials.
Work completed before a later error remains completed. Give knowledge writes and notifications stable idempotency_key values so retries do not duplicate them.
The runtime records success, failure, logs, and SDK calls on the behavior run.

When to use a reaction

Need	Use a reaction?
Persist the behavior’s normal extracted output	No. Lobu already records it.
Save an additional derived event	Yes.
Update or link entities deterministically	Yes.
Notify people only when a condition is met	Yes.
Execute a governed connector operation	Yes.
Change how the agent reasons about the source data	No. Put that in the behavior prompt or a skill.

Author a connector

Install

A typed connector, end to end

Concepts

ConnectorDefinition

SyncContext

EventEnvelope

SyncResult

ActionContext / ActionResult

Auth models

Checkpoints

Where the file lives

Dependencies

See it in production

Reactions

Install

A complete reaction

Attach it to a behavior

What the client can do

Save knowledge

Update an entity

Execute a connector operation

Reaction context

Runtime behavior

When to use a reaction

See also

Keep exploring

`ConnectorDefinition`

`SyncContext`

`EventEnvelope`

`SyncResult`

`ActionContext` / `ActionResult`