You are helping build a connector for Omni, an open-source AI agent platform for the workplace. Connectors are lightweight bridges between Omni and third-party systems: they sync workplace data into Omni's index and expose safe actions, MCP tools, resources, and prompts that workplace agents can use.
Core Principles
- One container per connector. Independently packaged and deployed.
- Connectors never interact directly with Omni services. Everything goes through connector-manager via the SDK.
- SDKs are local dependencies in this monorepo, not published package registries.
- Python, TypeScript, and Rust are first-class. Pick the language that best fits the provider SDK/ecosystem.
- Agents depend on connector contracts. Model read/write actions, permissions, OAuth scopes, resources, prompts, and document IDs carefully.
Scheduled sync flow: connector-manager creates a sync run -> sends POST /sync to the connector -> connector fetches source config/credentials through the SDK -> connector fetches provider data -> stores/extracts content via SDK -> emits events -> checkpoints as it advances -> completes or fails through SDK.
Realtime sync flow: connector-manager starts/monitors a long-running realtime sync when the manifest declares support -> connector keeps a webhook/socket/watcher alive, heartbeats while idle, and emits repair/update events or creates short follow-up sync runs as needed.
Checklist
Every new built-in connector usually requires changes across these areas:
- Connector implementation — sync logic, provider client, config/credential types, manifest, actions/tools/resources.
- Database migration — add source type to
sources_source_type_check; add service provider to service_credentials_provider_check if new.
- Rust
SourceType enum — add variant in shared/src/models.rs even for Python/TS connectors, because connector-manager is Rust.
- Frontend —
SourceType, ServiceProvider if new, icon, setup dialog, integrations page, sync interval defaults.
- Docker Compose — service definition, dev override, port in
.env.example, ENABLED_CONNECTORS comment.
- Terraform — AWS ECS task/service and GCP Cloud Run service.
- GitHub Actions — path filter, build job, release matrix entry.
- Integration tests — mock provider API, connector-manager/testcontainer harness, sync/action assertions.
Choosing a Language
|
Python |
TypeScript |
Rust |
| SDK |
sdk/python/ |
sdk/typescript/ |
sdk/rust/ crate omni-connector-sdk |
| Abstraction |
Connector base class, FastAPI server/registration generated |
Connector<TConfig, TCredentials, TState> base class, Express server/registration generated |
Connector trait, Axum server/registration generated by SDK |
| Typical sync state |
dict[str, Any] unless you define/parse concrete models |
generic TState type |
associated type State: DeserializeOwned + Serialize |
| Test harness |
omni_connector.testing |
SDK unit tests; no full connector harness yet |
Cargo integration tests with connector-manager/testcontainers |
| Reference connectors |
connectors/notion/, connectors/github/, connectors/clickup/, connectors/microsoft/ |
connectors/linear/ |
connectors/google/, connectors/slack/, connectors/atlassian/, connectors/filesystem/, connectors/nextcloud/ |
| Simple example |
sdk/python/examples/rss_connector.py |
sdk/typescript/examples/rss-connector.ts |
connectors/filesystem/ for a compact Rust SDK connector |
When implementing, read reference connectors in the chosen language. They are the canonical examples of project structure, Dockerfile/package config, SDK usage, and sync/action patterns.
Connector Protocol
The SDK server exposes these HTTP endpoints. Python/TypeScript/Rust SDKs generate them automatically unless you intentionally add extra routes in Rust with serve_with_extra_routes.
GET /health — health check.
GET /manifest — connector capabilities and agent-facing contract.
GET /sync/{sync_run_id} — whether this connector process still has the sync in memory; connector-manager uses this to detect lost syncs and auto-resume.
POST /sync — trigger or resume a sync. Payload includes sync_run_id, source_id, sync_mode, optional checkpoint, and is_resume.
POST /cancel — cancel a running sync.
POST /action — execute a connector action/tool. Connector-manager proxies the connector response status, headers, and body.
POST /resource — read an MCP resource exposed by the connector.
POST /prompt — get an MCP prompt exposed by the connector.
The connector auto-registers its manifest with connector-manager every 30 seconds. Registered manifests have a short TTL; connector-manager treats unregistered connectors as unavailable.
Manifest
The manifest describes both sync capability and agent capability. Key fields:
name, display_name, version, description
connector_id, connector_url
source_types — must match shared/src/models.rs SourceType and frontend SourceType.
sync_modes — any subset of full, incremental, realtime that is truly implemented.
search_operators
actions
read_only — Rust SDK supports connector-level read-only; connector-manager blocks write actions for read-only connectors/sources.
extra_schema, attributes_schema
mcp_enabled, resources, prompts
oauth — declarative OAuth2 config when the connector supports user-delegated OAuth.
Search Operators
Search operators let users filter results with keywords like from:alice@example.com.
operator — keyword users type, e.g. from, channel, status.
attribute_key — document attribute to filter on, e.g. sender, channel_name.
value_type — usually person, text, or datetime.
For operators to work, emit matching keys in each document's attributes. Attributes are for filtering/faceting/tool routing; they are not embedded as document body text.
Content Storage
Store content through connector-manager. Do not write directly to storage tables.
save(content, content_type) — text/HTML/Markdown, returns content_id.
extract_and_store_content(data, mime_type, filename) — binary files; connector-manager extracts text via Docling when enabled or built-in extractors, stores the extracted text, returns content_id.
extract_text(data, mime_type, filename) — same extraction but returns text without storing; use when you need to combine/post-process text before saving.
save_binary(content, content_type) — raw binary as base64, returns content_id (Python/TS helper; Rust can store text/extracted content directly).
Prefer extract_and_store_content for user-facing files and attachments. If extraction fails for a document but metadata is still useful, emit a metadata-only/fallback text document rather than silently dropping it.
Emitting Documents and Events
Document fields:
external_id — stable provider ID. This is what actions usually need after connector-manager resolves an Omni document ID.
title
content_id
metadata — title, author, created_at, updated_at, url, mime_type, size, path, content_type, extra.
permissions — public, users, groups.
attributes — filter/facet/tool-routing metadata, not embedded body text.
Event types:
document_created
document_updated
document_deleted
group_membership_sync
Use idempotent external_id/document_id values. Resume and retries can re-emit already completed units; indexer upserts/deletes should make that safe.
Sync Modes, Checkpointing, and Resume
Always branch on the dispatched sync mode, not on whether state/checkpoint exists. A manual full sync can arrive with an existing checkpoint and must still perform full-crawl/full-reconciliation behavior. Connector-manager may upgrade the first scheduled incremental request for a source to full when there is no prior completed sync; honor the mode you receive.
Full Sync
- Crawl all in-scope containers/items.
- Decide whether full sync should reset provider cursors, reconcile deletes, and rebuild permission groups.
- For long crawls, checkpoint after each durable unit: user/mailbox, workspace, folder, channel, project, page-token range, etc.
- Do not skip a unit on resume unless the checkpoint proves it completed after all its events were flushed.
Incremental Sync
- Prefer provider-native change cursors/history APIs.
- If only timestamps exist, use an overlap window and idempotent upserts/deletes to tolerate clock skew, late updates, and provider ordering quirks.
- Never advance a cursor/watermark before all items covered by it have been stored, emitted, and flushed.
- Handle cursor expiry by falling back to an appropriate full or scoped repair sync.
Realtime Sync
Only declare realtime when the provider supports a reliable event path: webhooks, socket mode, file watchers, subscriptions, or equivalent.
Realtime connectors must:
- Keep the long-running watcher healthy and call
heartbeat() while idle.
- Stop promptly when
ctx.is_cancelled() flips.
- Reconcile event gaps with periodic incremental/full repair.
- Renew/verify provider subscriptions when applicable.
- Store subscription IDs, expirations, and last processed event times in source-level connector metadata when the SDK exposes that safely.
- Use
document_updated/document_deleted for repair events when possible.
Current Rust SDK behavior to know:
SyncType::Realtime occupies a separate per-source slot from Full/Incremental, so a realtime watcher does not block scheduled scans.
- Rust SDK auto-completes non-realtime syncs when
Connector::sync() returns Ok(()); it does not auto-complete realtime syncs.
- For optional realtime support, implement
validate_sync_request() and return SyncRequestValidationError::Unavailable(...) when prerequisites are missing. Connector-manager treats a realtime 404/unavailable as “not available”, not as a broken connector.
Checkpoint and Resume Contract
Connector-manager maintains two different pieces of state:
- Run checkpoint (
sync_runs.checkpoint) — saved by save_checkpoint; used to resume the same running sync after connector restart/loss.
- Source checkpoint (
sources.checkpoint) — latest successful checkpoint; promoted from the run checkpoint only when the sync completes successfully.
Resume behavior:
- Fresh sync gets
checkpoint = source.checkpoint and is_resume = false.
- Auto-resume gets
checkpoint = sync_run.checkpoint if present, otherwise source.checkpoint, and is_resume = true.
- On
is_resume=true, continue the same logical run: do not reset full-sync progress, do not advance provider cursors optimistically, and reprocess any in-progress unit idempotently.
- Failed/cancelled run checkpoints are not promoted to the source.
- Completion atomically marks the run completed and publishes the run checkpoint to the source.
Checkpointing rules:
save_checkpoint(...) flushes buffered events before persisting the checkpoint in all SDKs. Use it after completed durable units.
- Never checkpoint past buffered/unflushed events.
- On resume, only skip units proven complete. Reprocess unfinished units idempotently.
- Poll cancellation in loops.
- Keep heartbeats fresh during long provider calls/page loops; checkpoint or explicit heartbeat is enough.
- Update progress counters as work completes. Use
increment_scanned / incrementScanned / increment_scanned(count) for scanned items. In Rust and TypeScript, call increment_updated / incrementUpdated when you need the manager-side updated count to survive crashes.
Language-Specific State APIs
| Concern |
Python |
TypeScript |
Rust |
| Dispatched mode |
ctx.sync_mode (SyncMode) |
ctx.syncMode (SyncMode) |
ctx.sync_mode() (SyncType) |
| Resume flag |
ctx.is_resume |
ctx.isResume |
ctx.is_resume() |
| Current checkpoint |
checkpoint arg and ctx.checkpoint |
state arg and ctx.state |
state: Option<Self::State> arg |
| Save checkpoint |
await ctx.save_checkpoint({...}) |
await ctx.saveCheckpoint({...}) / saveState alias |
ctx.save_checkpoint(json).await? |
| Complete |
await ctx.complete(checkpoint={...}) |
await ctx.complete({...}) |
normally return Ok(()) after saving checkpoint; ctx.complete().await? is also available |
| Cancel check |
ctx.is_cancelled() |
ctx.isCancelled() |
ctx.is_cancelled() |
| Mark cancelled |
no public cancel helper; current connectors generally stop/fail |
no public cancel helper |
ctx.cancel().await? |
| Source-level connector metadata |
ctx.connector_state, ctx.save_connector_state(...) |
SdkClient.updateConnectorState(...); SyncContext does not expose a helper today |
SdkClient::get_connector_state / save_connector_state; do not use deprecated SyncContext::save_connector_state for source metadata because it aliases checkpoint |
Actions and Agent Tools
Actions are how connectors expose provider capabilities to Omni agents and setup/UI flows. Define them deliberately.
ActionDefinition fields:
name — stable action name.
description — agent-facing description; explain when to use it and important IDs/side effects.
input_schema — JSON Schema object for parameters. Keep it strict and concrete.
mode — read or write. Default is write in shared/Rust, so set read explicitly for non-mutating actions.
source_types — restrict action to specific source types. Empty means all source types for that connector.
admin_only — connector-manager resolves org/admin credentials, not per-user OAuth credentials.
hidden — hidden from all chat/agent tool listings but still present in the manifest and dispatchable by name for setup/internal flows. Python/Rust support this; TypeScript currently may need SDK updates before setting it.
Connector-manager behavior:
/actions excludes hidden actions and write actions blocked by connector/source read_only.
execute_action blocks write actions for read-only connectors or sources.
- User-scoped agent calls require per-user credentials unless the action is
admin_only; when missing, connector-manager returns 412 needs_user_auth with an OAuth start URL.
- Source config is merged into action params, with caller params taking precedence.
document_id / file_id params may be resolved from Omni document ID to provider external_id before dispatch.
- Connector-manager proxies the connector's response status, headers, and body. Actions may return JSON or binary/file responses.
Action implementation rules:
- Declaring manual
actions only advertises them; override execute_action / executeAction / execute_action(...) to implement dispatch. The base classes only handle MCP tools and otherwise return not supported.
- Validate required params and credential shape immediately.
- Return
ActionResponse.success(...), failure(...), or not_supported(...) for JSON actions.
- For file/download actions, return a real HTTP response with content type and filename headers.
- Prefer source-native IDs in action params, but document in the schema when Omni document IDs are accepted/resolved.
- Keep write actions narrow and safe; mark
mode="write" and require user OAuth where appropriate.
MCP, Resources, Prompts, and OAuth
SDKs can bridge an external MCP server into the connector protocol.
To enable MCP:
- Override
mcp_server / mcp_server() with stdio or Streamable HTTP config.
- Implement
prepare_mcp_env(...) for stdio auth or prepare_mcp_headers(...) for HTTP auth.
- SDK bootstrap discovers MCP tools/resources/prompts after credentials are available.
- MCP tools are merged into manifest
actions; MCP resources/prompts populate manifest resources/prompts and are served through /resource and /prompt.
- MCP read-only hints are converted into action
mode=read; otherwise tools default to write.
OAuth manifest support:
- Python/Rust expose
oauth_config() returning OAuthManifestConfig.
- Declare provider auth/token/userinfo endpoints, identity scopes, per-source-type read/write scopes, extra auth params, scope separator, and optional enrich endpoint.
- Use read scopes for sync and read actions; use narrower write scopes for write tools.
- For user-context actions, connector-manager resolves per-user credentials and prompts for OAuth when missing.
Permissions and Inheritance
- Model the provider's authorization semantics before choosing document granularity.
- Identify whether permissions are item-level, container-inherited, mailbox-owned, workspace-wide, or group-derived.
- Emit
permissions.users and permissions.groups using stable identifiers already understood by Omni's permission filter.
- If permissions inherit from a parent container, store parent/container IDs in metadata/attributes and update affected child documents when membership changes.
- If the provider supports group principals, emit
group_membership_sync events where possible.
- Be conservative with private/DM/personal content. Prefer allowlists and opt-in handling until ACL semantics are proven.
- Use
ctx.should_index_user(...) / ctx.shouldIndexUser(...) where available before emitting per-user records under source whitelist/blacklist settings.
Document Modeling, Threads, and Attachments
- Use stable provider IDs for
external_id; avoid per-crawl or user-local IDs unless the source truly has user-local objects.
- Decide document granularity deliberately: item, message, thread, day/channel batch, file attachment, etc. This affects dedupe, incremental sync, permissions, snippets, and retrieval.
- Preserve parent references in metadata/attributes:
container_id, thread_id, parent_message_id, attachment_id, etc.
- For threaded systems, decide whether to index each message separately, group by thread, or re-emit the full thread on changes. Implement incremental/realtime repair accordingly.
- For attachments, distinguish linked provider files from uploaded blobs. Prefer extracting/storing attachment content as its own document when independently useful, with a back-reference to the parent item.
- For unsupported/oversized/private attachments, emit metadata-only documents or pointers rather than silently dropping them.
Frontend Changes
Read existing connector patterns before editing. Typical files:
web/src/lib/types.ts — SourceType, ServiceProvider if new, source config interface, DEFAULT_SYNC_INTERVAL_SECONDS.
web/src/lib/utils/icons.ts — register SVG icon and display name.
web/src/lib/images/icons/ — icon asset.
web/src/lib/components/*-setup.svelte — setup dialog component.
web/src/routes/(admin)/admin/settings/integrations/+page.svelte — import/render setup component.
web/src/routes/(admin)/admin/settings/integrations/+page.server.ts — CONNECTOR_DISPLAY_ORDER.
Icon Sourcing
- Fetch a local SVG icon for the app/service; do not hotlink remote assets.
- Prefer Wikimedia Commons when available because it usually has stable SVG files and explicit license metadata. Use the original SVG file URL, not a rendered PNG thumbnail.
- If Wikimedia does not have a suitable official mark, use the provider's official brand/media kit or another reputable source with compatible licensing. Record the source URL in your summary/PR notes.
- Verify the asset license before committing. Do not add icons with unclear, incompatible, or non-redistributable terms.
- Save the SVG under
web/src/lib/images/icons/ using the existing naming style, then register it in web/src/lib/utils/icons.ts.
- Keep the SVG compact and safe: no scripts, external references, embedded rasters, tracking metadata, or remote font/image links.
Docker Compose and ENABLED_CONNECTORS
Add connector service to docker/docker-compose.yml and dev overrides to docker/docker-compose.dev.yml. Follow existing connectors.
Add a port variable to .env.example and update the ENABLED_CONNECTORS comment.
ENABLED_CONNECTORS maps directly to Docker Compose profiles:
ENABLED_CONNECTORS=google,slack,my-connector
COMPOSE_PROFILES=${ENABLED_CONNECTORS}
Only containers whose profiles: value appears in this list start. Core services run regardless.
Terraform
Follow existing connector patterns:
- AWS (
infra/aws/terraform/modules/compute/): add task definition and ECS service in task_definitions.tf and services.tf, gated by count = contains(var.enabled_connectors, "name") ? 1 : 0.
- GCP (
infra/gcp/terraform/modules/compute/services.tf): add to all_simple_connectors for simple connectors, or add a count-based Cloud Run resource for complex ones.
Integration Testing
Prefer integration tests with real connector-manager/Postgres/Redis infrastructure.
Python harness at sdk/python/omni_connector/testing/ provides ParadeDB/Postgres, Redis, and connector-manager via testcontainers. See connectors/notion/tests/, connectors/github/tests/, and connectors/clickup/tests/.
Typical Python pattern:
- Mock third-party API with a Starlette app in a daemon thread.
- Start connector server in a daemon thread.
- Session-scoped
OmniTestHarness starts infra + connector-manager.
- Function-scoped fixtures seed source/credentials pointing to mock API.
- Trigger sync via connector-manager
POST /sync.
- Assert with
wait_for_sync(), count_events(), get_events(), source checkpoint helpers, and action calls.
Rust connectors use Cargo integration tests with connector-manager/testcontainers; see connectors/slack/tests/, connectors/web/tests/, and connectors/atlassian/tests/.
Test these behaviors when relevant:
- full sync from empty state
- manual full sync when a checkpoint already exists
- incremental sync from checkpoint
- checkpoint promotion only on success
- resume with
is_resume=true and run checkpoint
- cancellation
- action schema/listing/dispatch, including hidden/admin/read-only behavior
- permissions/group membership
- realtime watcher heartbeat/cancel/unavailable path
Run examples:
cd connectors/my-python-connector && uv run pytest
cargo test -p omni-my-connector
Key Files Reference
Connector SDKs
| What |
Python |
TypeScript |
Rust |
| Base class / trait |
sdk/python/omni_connector/connector.py |
sdk/typescript/src/connector.ts |
sdk/rust/src/connector.rs |
| Sync context |
sdk/python/omni_connector/context.py |
sdk/typescript/src/context.ts |
sdk/rust/src/context.rs |
| SDK client |
sdk/python/omni_connector/client.py |
sdk/typescript/src/client.ts |
sdk/rust/src/client.rs |
| Content storage |
sdk/python/omni_connector/storage.py |
sdk/typescript/src/storage.ts |
methods on Rust SyncContext/SdkClient |
| Data models |
sdk/python/omni_connector/models.py |
sdk/typescript/src/models.ts |
sdk/rust/src/models.rs + shared/src/models.rs |
| Server / registration |
sdk/python/omni_connector/server.py |
sdk/typescript/src/server.ts |
sdk/rust/src/server.rs |
| MCP adapter |
sdk/python/omni_connector/mcp_adapter.py |
sdk/typescript/src/mcp-adapter.ts |
sdk/rust/src/mcp_adapter.rs |
| Simple example |
sdk/python/examples/rss_connector.py |
sdk/typescript/examples/rss-connector.ts |
connectors/filesystem/ |
Backend Shared Files
| What |
File |
| Source types, manifests, action definitions, sync types/events |
shared/src/models.rs |
| Connector-manager sync orchestration/resume |
services/connector-manager/src/sync_manager.rs |
| Connector-manager action/resource/prompt dispatch |
services/connector-manager/src/handlers.rs |
| Connector HTTP client |
services/connector-manager/src/connector_client.rs |
| Scheduler and realtime startup |
services/connector-manager/src/scheduler.rs |
| Migration pattern for source type |
services/migrations/075_add_paperless_ngx_source_type.sql or latest source-type migration |
| Migration directory |
services/migrations/ |
Frontend
| What |
File |
| Source type & provider enums |
web/src/lib/types.ts |
| Icons & display names |
web/src/lib/utils/icons.ts |
| Icon assets |
web/src/lib/images/icons/ |
| Setup dialog examples |
web/src/lib/components/*-setup.svelte |
| Integrations page |
web/src/routes/(admin)/admin/settings/integrations/+page.svelte |
| Display order |
web/src/routes/(admin)/admin/settings/integrations/+page.server.ts |
Infrastructure and CI
| What |
File |
| Docker Compose services |
docker/docker-compose.yml |
| Docker Compose dev overrides |
docker/docker-compose.dev.yml |
| Port assignments & ENABLED_CONNECTORS |
.env.example |
| AWS task definitions |
infra/aws/terraform/modules/compute/task_definitions.tf |
| AWS ECS services |
infra/aws/terraform/modules/compute/services.tf |
| GCP Cloud Run |
infra/gcp/terraform/modules/compute/services.tf |
| Path filters/build/release matrix |
.github/workflows/ci.yml |
Testing
| What |
File |
| Python test harness |
sdk/python/omni_connector/testing/harness.py |
| Python DB seeding |
sdk/python/omni_connector/testing/seed.py |
| Python assertions |
sdk/python/omni_connector/testing/assertions.py |
| Python examples |
connectors/notion/tests/, connectors/github/tests/, connectors/clickup/tests/ |
| Rust examples |
connectors/slack/tests/, connectors/web/tests/, connectors/atlassian/tests/ |
Coding Guidelines
- Use concrete types, not
dict[str, Any] / Record<string, unknown> / serde_json::Value, when the shape is known. Opaque maps are acceptable only at SDK boundaries or for genuinely provider-dynamic payloads.
- No empty string
"" for missing state; use None / null / Option.
- Fail immediately on missing required config/credentials.
- Imports at the top of the file.
- Only add comments explaining why, not what.
- Prefer integration tests with real infrastructure over isolated mocks.
- Python: use
uv, not pip, for connector development in this repo.
- Svelte buttons must use Tailwind
cursor-pointer.
1---2name: build-connector3description: Build a new connector for Omni workplace agents. Use when creating a new data source integration, scaffolding connector structure, exposing agent actions/tools/resources, or needing help with the connector SDK.4---56You are helping build a connector for Omni, an open-source AI agent platform for the workplace. Connectors are lightweight bridges between Omni and third-party systems: they sync workplace data into Omni's index and expose safe actions, MCP tools, resources, and prompts that workplace agents can use.78# Core Principles910- **One container per connector.** Independently packaged and deployed.11- **Connectors never interact directly with Omni services.** Everything goes through connector-manager via the SDK.12- **SDKs are local dependencies** in this monorepo, not published package registries.13- **Python, TypeScript, and Rust are first-class.** Pick the language that best fits the provider SDK/ecosystem.14- **Agents depend on connector contracts.** Model read/write actions, permissions, OAuth scopes, resources, prompts, and document IDs carefully.1516**Scheduled sync flow:** connector-manager creates a sync run -> sends `POST /sync` to the connector -> connector fetches source config/credentials through the SDK -> connector fetches provider data -> stores/extracts content via SDK -> emits events -> checkpoints as it advances -> completes or fails through SDK.1718**Realtime sync flow:** connector-manager starts/monitors a long-running `realtime` sync when the manifest declares support -> connector keeps a webhook/socket/watcher alive, heartbeats while idle, and emits repair/update events or creates short follow-up sync runs as needed.1920# Checklist2122Every new built-in connector usually requires changes across these areas:23241. **Connector implementation** — sync logic, provider client, config/credential types, manifest, actions/tools/resources.252. **Database migration** — add source type to `sources_source_type_check`; add service provider to `service_credentials_provider_check` if new.263. **Rust `SourceType` enum** — add variant in `shared/src/models.rs` even for Python/TS connectors, because connector-manager is Rust.274. **Frontend** — `SourceType`, `ServiceProvider` if new, icon, setup dialog, integrations page, sync interval defaults.285. **Docker Compose** — service definition, dev override, port in `.env.example`, `ENABLED_CONNECTORS` comment.296. **Terraform** — AWS ECS task/service and GCP Cloud Run service.307. **GitHub Actions** — path filter, build job, release matrix entry.318. **Integration tests** — mock provider API, connector-manager/testcontainer harness, sync/action assertions.3233# Choosing a Language3435| | Python | TypeScript | Rust |36|---|---|---|---|37| **SDK** | `sdk/python/` | `sdk/typescript/` | `sdk/rust/` crate `omni-connector-sdk` |38| **Abstraction** | `Connector` base class, FastAPI server/registration generated | `Connector<TConfig, TCredentials, TState>` base class, Express server/registration generated | `Connector` trait, Axum server/registration generated by SDK |39| **Typical sync state** | `dict[str, Any]` unless you define/parse concrete models | generic `TState` type | associated `type State: DeserializeOwned + Serialize` |40| **Test harness** | `omni_connector.testing` | SDK unit tests; no full connector harness yet | Cargo integration tests with connector-manager/testcontainers |41| **Reference connectors** | `connectors/notion/`, `connectors/github/`, `connectors/clickup/`, `connectors/microsoft/` | `connectors/linear/` | `connectors/google/`, `connectors/slack/`, `connectors/atlassian/`, `connectors/filesystem/`, `connectors/nextcloud/` |42| **Simple example** | `sdk/python/examples/rss_connector.py` | `sdk/typescript/examples/rss-connector.ts` | `connectors/filesystem/` for a compact Rust SDK connector |4344When implementing, read reference connectors in the chosen language. They are the canonical examples of project structure, Dockerfile/package config, SDK usage, and sync/action patterns.4546# Connector Protocol4748The SDK server exposes these HTTP endpoints. Python/TypeScript/Rust SDKs generate them automatically unless you intentionally add extra routes in Rust with `serve_with_extra_routes`.4950- `GET /health` — health check.51- `GET /manifest` — connector capabilities and agent-facing contract.52- `GET /sync/{sync_run_id}` — whether this connector process still has the sync in memory; connector-manager uses this to detect lost syncs and auto-resume.53- `POST /sync` — trigger or resume a sync. Payload includes `sync_run_id`, `source_id`, `sync_mode`, optional `checkpoint`, and `is_resume`.54- `POST /cancel` — cancel a running sync.55- `POST /action` — execute a connector action/tool. Connector-manager proxies the connector response status, headers, and body.56- `POST /resource` — read an MCP resource exposed by the connector.57- `POST /prompt` — get an MCP prompt exposed by the connector.5859The connector auto-registers its manifest with connector-manager every 30 seconds. Registered manifests have a short TTL; connector-manager treats unregistered connectors as unavailable.6061# Manifest6263The manifest describes both sync capability and agent capability. Key fields:6465- `name`, `display_name`, `version`, `description`66- `connector_id`, `connector_url`67- `source_types` — must match `shared/src/models.rs` `SourceType` and frontend `SourceType`.68- `sync_modes` — any subset of `full`, `incremental`, `realtime` that is truly implemented.69- `search_operators`70- `actions`71- `read_only` — Rust SDK supports connector-level read-only; connector-manager blocks write actions for read-only connectors/sources.72- `extra_schema`, `attributes_schema`73- `mcp_enabled`, `resources`, `prompts`74- `oauth` — declarative OAuth2 config when the connector supports user-delegated OAuth.7576## Search Operators7778Search operators let users filter results with keywords like `from:alice@example.com`.7980- `operator` — keyword users type, e.g. `from`, `channel`, `status`.81- `attribute_key` — document attribute to filter on, e.g. `sender`, `channel_name`.82- `value_type` — usually `person`, `text`, or `datetime`.8384For operators to work, emit matching keys in each document's `attributes`. Attributes are for filtering/faceting/tool routing; they are not embedded as document body text.8586# Content Storage8788Store content through connector-manager. Do not write directly to storage tables.89901. **`save(content, content_type)`** — text/HTML/Markdown, returns `content_id`.912. **`extract_and_store_content(data, mime_type, filename)`** — binary files; connector-manager extracts text via Docling when enabled or built-in extractors, stores the extracted text, returns `content_id`.923. **`extract_text(data, mime_type, filename)`** — same extraction but returns text without storing; use when you need to combine/post-process text before saving.934. **`save_binary(content, content_type)`** — raw binary as base64, returns `content_id` (Python/TS helper; Rust can store text/extracted content directly).9495Prefer `extract_and_store_content` for user-facing files and attachments. If extraction fails for a document but metadata is still useful, emit a metadata-only/fallback text document rather than silently dropping it.9697# Emitting Documents and Events9899Document fields:100101- `external_id` — stable provider ID. This is what actions usually need after connector-manager resolves an Omni document ID.102- `title`103- `content_id`104- `metadata` — `title`, `author`, `created_at`, `updated_at`, `url`, `mime_type`, `size`, `path`, `content_type`, `extra`.105- `permissions` — `public`, `users`, `groups`.106- `attributes` — filter/facet/tool-routing metadata, not embedded body text.107108Event types:109110- `document_created`111- `document_updated`112- `document_deleted`113- `group_membership_sync`114115Use idempotent `external_id`/`document_id` values. Resume and retries can re-emit already completed units; indexer upserts/deletes should make that safe.116117# Sync Modes, Checkpointing, and Resume118119Always branch on the **dispatched sync mode**, not on whether state/checkpoint exists. A manual full sync can arrive with an existing checkpoint and must still perform full-crawl/full-reconciliation behavior. Connector-manager may upgrade the first scheduled incremental request for a source to `full` when there is no prior completed sync; honor the mode you receive.120121## Full Sync122123- Crawl all in-scope containers/items.124- Decide whether full sync should reset provider cursors, reconcile deletes, and rebuild permission groups.125- For long crawls, checkpoint after each durable unit: user/mailbox, workspace, folder, channel, project, page-token range, etc.126- Do not skip a unit on resume unless the checkpoint proves it completed after all its events were flushed.127128## Incremental Sync129130- Prefer provider-native change cursors/history APIs.131- If only timestamps exist, use an overlap window and idempotent upserts/deletes to tolerate clock skew, late updates, and provider ordering quirks.132- Never advance a cursor/watermark before all items covered by it have been stored, emitted, and flushed.133- Handle cursor expiry by falling back to an appropriate full or scoped repair sync.134135## Realtime Sync136137Only declare `realtime` when the provider supports a reliable event path: webhooks, socket mode, file watchers, subscriptions, or equivalent.138139Realtime connectors must:140141- Keep the long-running watcher healthy and call `heartbeat()` while idle.142- Stop promptly when `ctx.is_cancelled()` flips.143- Reconcile event gaps with periodic incremental/full repair.144- Renew/verify provider subscriptions when applicable.145- Store subscription IDs, expirations, and last processed event times in source-level connector metadata when the SDK exposes that safely.146- Use `document_updated`/`document_deleted` for repair events when possible.147148Current Rust SDK behavior to know:149150- `SyncType::Realtime` occupies a separate per-source slot from `Full`/`Incremental`, so a realtime watcher does not block scheduled scans.151- Rust SDK auto-completes non-realtime syncs when `Connector::sync()` returns `Ok(())`; it does **not** auto-complete realtime syncs.152- For optional realtime support, implement `validate_sync_request()` and return `SyncRequestValidationError::Unavailable(...)` when prerequisites are missing. Connector-manager treats a realtime 404/unavailable as “not available”, not as a broken connector.153154## Checkpoint and Resume Contract155156Connector-manager maintains two different pieces of state:157158- **Run checkpoint** (`sync_runs.checkpoint`) — saved by `save_checkpoint`; used to resume the same running sync after connector restart/loss.159- **Source checkpoint** (`sources.checkpoint`) — latest successful checkpoint; promoted from the run checkpoint only when the sync completes successfully.160161Resume behavior:162163- Fresh sync gets `checkpoint = source.checkpoint` and `is_resume = false`.164- Auto-resume gets `checkpoint = sync_run.checkpoint` if present, otherwise `source.checkpoint`, and `is_resume = true`.165- On `is_resume=true`, continue the same logical run: do not reset full-sync progress, do not advance provider cursors optimistically, and reprocess any in-progress unit idempotently.166- Failed/cancelled run checkpoints are not promoted to the source.167- Completion atomically marks the run completed and publishes the run checkpoint to the source.168169Checkpointing rules:170171- `save_checkpoint(...)` flushes buffered events before persisting the checkpoint in all SDKs. Use it after completed durable units.172- Never checkpoint past buffered/unflushed events.173- On resume, only skip units proven complete. Reprocess unfinished units idempotently.174- Poll cancellation in loops.175- Keep heartbeats fresh during long provider calls/page loops; checkpoint or explicit heartbeat is enough.176- Update progress counters as work completes. Use `increment_scanned` / `incrementScanned` / `increment_scanned(count)` for scanned items. In Rust and TypeScript, call `increment_updated` / `incrementUpdated` when you need the manager-side updated count to survive crashes.177178## Language-Specific State APIs179180| Concern | Python | TypeScript | Rust |181|---|---|---|---|182| Dispatched mode | `ctx.sync_mode` (`SyncMode`) | `ctx.syncMode` (`SyncMode`) | `ctx.sync_mode()` (`SyncType`) |183| Resume flag | `ctx.is_resume` | `ctx.isResume` | `ctx.is_resume()` |184| Current checkpoint | `checkpoint` arg and `ctx.checkpoint` | `state` arg and `ctx.state` | `state: Option<Self::State>` arg |185| Save checkpoint | `await ctx.save_checkpoint({...})` | `await ctx.saveCheckpoint({...})` / `saveState` alias | `ctx.save_checkpoint(json).await?` |186| Complete | `await ctx.complete(checkpoint={...})` | `await ctx.complete({...})` | normally return `Ok(())` after saving checkpoint; `ctx.complete().await?` is also available |187| Cancel check | `ctx.is_cancelled()` | `ctx.isCancelled()` | `ctx.is_cancelled()` |188| Mark cancelled | no public cancel helper; current connectors generally stop/fail | no public cancel helper | `ctx.cancel().await?` |189| Source-level connector metadata | `ctx.connector_state`, `ctx.save_connector_state(...)` | `SdkClient.updateConnectorState(...)`; `SyncContext` does not expose a helper today | `SdkClient::get_connector_state` / `save_connector_state`; do **not** use deprecated `SyncContext::save_connector_state` for source metadata because it aliases checkpoint |190191# Actions and Agent Tools192193Actions are how connectors expose provider capabilities to Omni agents and setup/UI flows. Define them deliberately.194195`ActionDefinition` fields:196197- `name` — stable action name.198- `description` — agent-facing description; explain when to use it and important IDs/side effects.199- `input_schema` — JSON Schema object for parameters. Keep it strict and concrete.200- `mode` — `read` or `write`. Default is write in shared/Rust, so set read explicitly for non-mutating actions.201- `source_types` — restrict action to specific source types. Empty means all source types for that connector.202- `admin_only` — connector-manager resolves org/admin credentials, not per-user OAuth credentials.203- `hidden` — hidden from all chat/agent tool listings but still present in the manifest and dispatchable by name for setup/internal flows. Python/Rust support this; TypeScript currently may need SDK updates before setting it.204205Connector-manager behavior:206207- `/actions` excludes hidden actions and write actions blocked by connector/source `read_only`.208- `execute_action` blocks write actions for read-only connectors or sources.209- User-scoped agent calls require per-user credentials unless the action is `admin_only`; when missing, connector-manager returns `412 needs_user_auth` with an OAuth start URL.210- Source config is merged into action params, with caller params taking precedence.211- `document_id` / `file_id` params may be resolved from Omni document ID to provider `external_id` before dispatch.212- Connector-manager proxies the connector's response status, headers, and body. Actions may return JSON or binary/file responses.213214Action implementation rules:215216- Declaring manual `actions` only advertises them; override `execute_action` / `executeAction` / `execute_action(...)` to implement dispatch. The base classes only handle MCP tools and otherwise return not supported.217- Validate required params and credential shape immediately.218- Return `ActionResponse.success(...)`, `failure(...)`, or `not_supported(...)` for JSON actions.219- For file/download actions, return a real HTTP response with content type and filename headers.220- Prefer source-native IDs in action params, but document in the schema when Omni document IDs are accepted/resolved.221- Keep write actions narrow and safe; mark `mode="write"` and require user OAuth where appropriate.222223# MCP, Resources, Prompts, and OAuth224225SDKs can bridge an external MCP server into the connector protocol.226227To enable MCP:228229- Override `mcp_server` / `mcp_server()` with stdio or Streamable HTTP config.230- Implement `prepare_mcp_env(...)` for stdio auth or `prepare_mcp_headers(...)` for HTTP auth.231- SDK bootstrap discovers MCP tools/resources/prompts after credentials are available.232- MCP tools are merged into manifest `actions`; MCP resources/prompts populate manifest `resources`/`prompts` and are served through `/resource` and `/prompt`.233- MCP read-only hints are converted into action `mode=read`; otherwise tools default to write.234235OAuth manifest support:236237- Python/Rust expose `oauth_config()` returning `OAuthManifestConfig`.238- Declare provider auth/token/userinfo endpoints, identity scopes, per-source-type read/write scopes, extra auth params, scope separator, and optional enrich endpoint.239- Use read scopes for sync and read actions; use narrower write scopes for write tools.240- For user-context actions, connector-manager resolves per-user credentials and prompts for OAuth when missing.241242# Permissions and Inheritance243244- Model the provider's authorization semantics before choosing document granularity.245- Identify whether permissions are item-level, container-inherited, mailbox-owned, workspace-wide, or group-derived.246- Emit `permissions.users` and `permissions.groups` using stable identifiers already understood by Omni's permission filter.247- If permissions inherit from a parent container, store parent/container IDs in metadata/attributes and update affected child documents when membership changes.248- If the provider supports group principals, emit `group_membership_sync` events where possible.249- Be conservative with private/DM/personal content. Prefer allowlists and opt-in handling until ACL semantics are proven.250- Use `ctx.should_index_user(...)` / `ctx.shouldIndexUser(...)` where available before emitting per-user records under source whitelist/blacklist settings.251252# Document Modeling, Threads, and Attachments253254- Use stable provider IDs for `external_id`; avoid per-crawl or user-local IDs unless the source truly has user-local objects.255- Decide document granularity deliberately: item, message, thread, day/channel batch, file attachment, etc. This affects dedupe, incremental sync, permissions, snippets, and retrieval.256- Preserve parent references in metadata/attributes: `container_id`, `thread_id`, `parent_message_id`, `attachment_id`, etc.257- For threaded systems, decide whether to index each message separately, group by thread, or re-emit the full thread on changes. Implement incremental/realtime repair accordingly.258- For attachments, distinguish linked provider files from uploaded blobs. Prefer extracting/storing attachment content as its own document when independently useful, with a back-reference to the parent item.259- For unsupported/oversized/private attachments, emit metadata-only documents or pointers rather than silently dropping them.260261# Frontend Changes262263Read existing connector patterns before editing. Typical files:2642651. **`web/src/lib/types.ts`** — `SourceType`, `ServiceProvider` if new, source config interface, `DEFAULT_SYNC_INTERVAL_SECONDS`.2662. **`web/src/lib/utils/icons.ts`** — register SVG icon and display name.2673. **`web/src/lib/images/icons/`** — icon asset.2684. **`web/src/lib/components/*-setup.svelte`** — setup dialog component.2695. **`web/src/routes/(admin)/admin/settings/integrations/+page.svelte`** — import/render setup component.2706. **`web/src/routes/(admin)/admin/settings/integrations/+page.server.ts`** — `CONNECTOR_DISPLAY_ORDER`.271272## Icon Sourcing273274- Fetch a local SVG icon for the app/service; do not hotlink remote assets.275- Prefer Wikimedia Commons when available because it usually has stable SVG files and explicit license metadata. Use the original SVG file URL, not a rendered PNG thumbnail.276- If Wikimedia does not have a suitable official mark, use the provider's official brand/media kit or another reputable source with compatible licensing. Record the source URL in your summary/PR notes.277- Verify the asset license before committing. Do not add icons with unclear, incompatible, or non-redistributable terms.278- Save the SVG under `web/src/lib/images/icons/` using the existing naming style, then register it in `web/src/lib/utils/icons.ts`.279- Keep the SVG compact and safe: no scripts, external references, embedded rasters, tracking metadata, or remote font/image links.280281# Docker Compose and ENABLED_CONNECTORS282283Add connector service to `docker/docker-compose.yml` and dev overrides to `docker/docker-compose.dev.yml`. Follow existing connectors.284285Add a port variable to `.env.example` and update the `ENABLED_CONNECTORS` comment.286287`ENABLED_CONNECTORS` maps directly to Docker Compose profiles:288289```env290ENABLED_CONNECTORS=google,slack,my-connector291COMPOSE_PROFILES=${ENABLED_CONNECTORS}292```293294Only containers whose `profiles:` value appears in this list start. Core services run regardless.295296# Terraform297298Follow existing connector patterns:299300- **AWS** (`infra/aws/terraform/modules/compute/`): add task definition and ECS service in `task_definitions.tf` and `services.tf`, gated by `count = contains(var.enabled_connectors, "name") ? 1 : 0`.301- **GCP** (`infra/gcp/terraform/modules/compute/services.tf`): add to `all_simple_connectors` for simple connectors, or add a count-based Cloud Run resource for complex ones.302303# Integration Testing304305Prefer integration tests with real connector-manager/Postgres/Redis infrastructure.306307Python harness at `sdk/python/omni_connector/testing/` provides ParadeDB/Postgres, Redis, and connector-manager via testcontainers. See `connectors/notion/tests/`, `connectors/github/tests/`, and `connectors/clickup/tests/`.308309Typical Python pattern:3103111. Mock third-party API with a Starlette app in a daemon thread.3122. Start connector server in a daemon thread.3133. Session-scoped `OmniTestHarness` starts infra + connector-manager.3144. Function-scoped fixtures seed source/credentials pointing to mock API.3155. Trigger sync via connector-manager `POST /sync`.3166. Assert with `wait_for_sync()`, `count_events()`, `get_events()`, source checkpoint helpers, and action calls.317318Rust connectors use Cargo integration tests with connector-manager/testcontainers; see `connectors/slack/tests/`, `connectors/web/tests/`, and `connectors/atlassian/tests/`.319320Test these behaviors when relevant:321322- full sync from empty state323- manual full sync when a checkpoint already exists324- incremental sync from checkpoint325- checkpoint promotion only on success326- resume with `is_resume=true` and run checkpoint327- cancellation328- action schema/listing/dispatch, including hidden/admin/read-only behavior329- permissions/group membership330- realtime watcher heartbeat/cancel/unavailable path331332Run examples:333334```bash335cd connectors/my-python-connector && uv run pytest336cargo test -p omni-my-connector337```338339# Key Files Reference340341## Connector SDKs342343| What | Python | TypeScript | Rust |344|---|---|---|---|345| Base class / trait | `sdk/python/omni_connector/connector.py` | `sdk/typescript/src/connector.ts` | `sdk/rust/src/connector.rs` |346| Sync context | `sdk/python/omni_connector/context.py` | `sdk/typescript/src/context.ts` | `sdk/rust/src/context.rs` |347| SDK client | `sdk/python/omni_connector/client.py` | `sdk/typescript/src/client.ts` | `sdk/rust/src/client.rs` |348| Content storage | `sdk/python/omni_connector/storage.py` | `sdk/typescript/src/storage.ts` | methods on Rust `SyncContext`/`SdkClient` |349| Data models | `sdk/python/omni_connector/models.py` | `sdk/typescript/src/models.ts` | `sdk/rust/src/models.rs` + `shared/src/models.rs` |350| Server / registration | `sdk/python/omni_connector/server.py` | `sdk/typescript/src/server.ts` | `sdk/rust/src/server.rs` |351| MCP adapter | `sdk/python/omni_connector/mcp_adapter.py` | `sdk/typescript/src/mcp-adapter.ts` | `sdk/rust/src/mcp_adapter.rs` |352| Simple example | `sdk/python/examples/rss_connector.py` | `sdk/typescript/examples/rss-connector.ts` | `connectors/filesystem/` |353354## Backend Shared Files355356| What | File |357|---|---|358| Source types, manifests, action definitions, sync types/events | `shared/src/models.rs` |359| Connector-manager sync orchestration/resume | `services/connector-manager/src/sync_manager.rs` |360| Connector-manager action/resource/prompt dispatch | `services/connector-manager/src/handlers.rs` |361| Connector HTTP client | `services/connector-manager/src/connector_client.rs` |362| Scheduler and realtime startup | `services/connector-manager/src/scheduler.rs` |363| Migration pattern for source type | `services/migrations/075_add_paperless_ngx_source_type.sql` or latest source-type migration |364| Migration directory | `services/migrations/` |365366## Frontend367368| What | File |369|---|---|370| Source type & provider enums | `web/src/lib/types.ts` |371| Icons & display names | `web/src/lib/utils/icons.ts` |372| Icon assets | `web/src/lib/images/icons/` |373| Setup dialog examples | `web/src/lib/components/*-setup.svelte` |374| Integrations page | `web/src/routes/(admin)/admin/settings/integrations/+page.svelte` |375| Display order | `web/src/routes/(admin)/admin/settings/integrations/+page.server.ts` |376377## Infrastructure and CI378379| What | File |380|---|---|381| Docker Compose services | `docker/docker-compose.yml` |382| Docker Compose dev overrides | `docker/docker-compose.dev.yml` |383| Port assignments & ENABLED_CONNECTORS | `.env.example` |384| AWS task definitions | `infra/aws/terraform/modules/compute/task_definitions.tf` |385| AWS ECS services | `infra/aws/terraform/modules/compute/services.tf` |386| GCP Cloud Run | `infra/gcp/terraform/modules/compute/services.tf` |387| Path filters/build/release matrix | `.github/workflows/ci.yml` |388389## Testing390391| What | File |392|---|---|393| Python test harness | `sdk/python/omni_connector/testing/harness.py` |394| Python DB seeding | `sdk/python/omni_connector/testing/seed.py` |395| Python assertions | `sdk/python/omni_connector/testing/assertions.py` |396| Python examples | `connectors/notion/tests/`, `connectors/github/tests/`, `connectors/clickup/tests/` |397| Rust examples | `connectors/slack/tests/`, `connectors/web/tests/`, `connectors/atlassian/tests/` |398399# Coding Guidelines400401- Use concrete types, not `dict[str, Any]` / `Record<string, unknown>` / `serde_json::Value`, when the shape is known. Opaque maps are acceptable only at SDK boundaries or for genuinely provider-dynamic payloads.402- No empty string `""` for missing state; use `None` / `null` / `Option`.403- Fail immediately on missing required config/credentials.404- Imports at the top of the file.405- Only add comments explaining why, not what.406- Prefer integration tests with real infrastructure over isolated mocks.407- Python: use `uv`, not `pip`, for connector development in this repo.408- Svelte buttons must use Tailwind `cursor-pointer`.