Python SDK

The Python SDK (omni-connector) provides everything you need to build custom connectors that integrate with Omni’s search and AI platform.

Installation

The SDK is not published to PyPI. Connectors live in the Omni monorepo and depend on the SDK as a local path dependency. To start a new Python connector, fork or clone omni and create a directory under connectors/:

git clone https://github.com/getomnico/omni.git
cd omni/connectors
mkdir my-connector && cd my-connector

Add a pyproject.toml that pulls the SDK from the local checkout — copy from any existing Python connector (e.g. connectors/notion/pyproject.toml):

[project]
name = "my-connector"
version = "1.0.0"
requires-python = ">=3.11"
dependencies = [
    "omni-connector",
]

[tool.uv.sources]
omni-connector = { path = "../../sdk/python" }

Then install with uv:

uv sync

Requirements:

Python >= 3.11
uv (used throughout the monorepo for Python dependency management)

Transitive dependencies pulled in by the SDK:

fastapi - Web framework
uvicorn - ASGI server
httpx - Async HTTP client
pydantic - Data validation

Run /build-connector <service name> from Claude Code inside the omni repo to scaffold the entire connector structure (sync logic, manifest, Dockerfile, frontend wiring, Terraform, integration tests) following the conventions used by built-in connectors. See the SDK overview for details.

Quick Start

from omni_connector import (
    Connector,
    Document,
    DocumentMetadata,
    DocumentPermissions,
    SyncContext,
)

class MyConnector(Connector):
    @property
    def name(self) -> str:
        return "my-connector"

    @property
    def version(self) -> str:
        return "1.0.0"

    @property
    def source_types(self) -> list[str]:
        return ["my_source_type"]

    @property
    def sync_modes(self) -> list[str]:
        return ["full", "incremental"]

    async def sync(
        self,
        source_config: dict,
        credentials: dict,
        checkpoint: dict | None,
        ctx: SyncContext
    ) -> None:
        # Your sync logic here
        await ctx.complete()

if __name__ == "__main__":
    MyConnector().serve(port=8000)

When running under Docker, set CONNECTOR_MANAGER_URL, CONNECTOR_HOST_NAME, and PORT. The SDK uses CONNECTOR_HOST_NAME plus PORT to register the connector URL that connector-manager calls for syncs, actions, MCP resources, and prompts.

Core Concepts

Connector Class

The Connector abstract base class defines the interface for all connectors. Required Properties:

Property	Type	Description
`name`	`str`	Unique connector identifier (e.g., `"my-connector"`)
`version`	`str`	Semantic version (e.g., `"1.0.0"`)
`source_types`	`list[str]`	Source types the connector handles

Optional Properties:

Property	Type	Default	Description
`sync_modes`	`list[str]`	`["full"]`	Supported modes: `"full"`, `"incremental"`
`actions`	`list[ActionDefinition]`	`[]`	Custom actions the connector supports. Set `hidden=True` on an action definition to keep internal/setup actions out of chat tools.
`skills`	`list[ConnectorSkillDefinition]`	`[]`	Connector-owned skills that guide the Omni agent through connector-specific workflows
`mcp_server`	`StdioMcpServer \| HttpMcpServer \| None`	`None`	MCP server that exposes tools to Omni

Required Methods:

async def sync(
    self,
    source_config: dict,      # Configuration from Omni UI
    credentials: dict,        # Stored credentials (API keys, tokens)
    checkpoint: dict | None,  # Latest successful or per-run checkpoint
    ctx: SyncContext          # Sync context utilities
) -> None:
    """Execute a sync operation."""
    pass

Optional Methods:

def cancel(self, sync_run_id: str) -> None:
    """Handle sync cancellation request."""
    pass

async def execute_action(
    self,
    action: str,
    params: dict,
    credentials: dict
) -> dict:
    """Execute a custom action."""
    pass

async def oauth_credential_ready(
    self,
    request: OAuthCredentialReadyRequest
) -> None:
    """Called after OAuth credentials are stored for a connector.

    Use this to run post-OAuth setup such as authenticated MCP
    catalog discovery."""
    pass

SyncContext

The SyncContext object is passed to your sync method and provides utilities for the sync operation. Properties:

Property	Type	Description
`sync_run_id`	`str`	Unique identifier for this sync run
`source_id`	`str`	Source identifier
`source_type`	`str \| None`	Source type dispatched by connector-manager
`checkpoint`	`dict`	Latest successful checkpoint, or the latest per-run checkpoint when resuming
`connector_state`	`dict`	Durable source-level metadata that is not used as the sync cursor
`sync_mode`	`SyncMode`	Dispatched mode: `full`, `incremental`, or `realtime`
`is_resume`	`bool`	Whether this run was resumed after an interruption
`content_storage`	`ContentStorage`	Storage for document content
`documents_emitted`	`int`	Count of emitted documents, seeded from the running sync on resume
`documents_scanned`	`int`	Count of scanned items, seeded from the running sync on resume

Methods:

# Emit a new document
await ctx.emit(document: Document)

# Emit an updated document
await ctx.emit_updated(document: Document)

# Mark a document as deleted
await ctx.emit_deleted(external_id: str)

# Emit source group membership for permission filtering
await ctx.emit_group_membership(
    group_email: str,
    member_emails: list[str],
    group_name: str | None = None,
)

# Report a non-fatal error for a document
await ctx.emit_error(external_id: str, error: str)

# Increment scanned counter (also sends heartbeat)
await ctx.increment_scanned()

# Check source-level user allow/deny settings before emitting user-owned docs
ctx.should_index_user(user_email: str) -> bool

# Save a per-run checkpoint for resumability
await ctx.save_checkpoint(checkpoint: dict)

# Persist source-level metadata that is not the resume cursor
await ctx.save_connector_state(connector_state: dict)

# Deprecated alias for save_checkpoint
await ctx.save_state(checkpoint: dict)

# Mark sync as completed, optionally saving the final checkpoint first
await ctx.complete(checkpoint: dict | None = None)

# Mark sync as failed
await ctx.fail(error: str)

# Check if sync was cancelled
ctx.is_cancelled() -> bool

ContentStorage

The ContentStorage class handles storing document content.

# Store text content
content_id = await ctx.content_storage.save(
    content: str,
    content_type: str = "text/plain"
) -> str

# Store binary content (base64 encoded)
content_id = await ctx.content_storage.save_binary(
    content: bytes,
    content_type: str
) -> str

# Extract text from a binary file (PDF, DOCX, PPTX, XLSX, images, ...)
# via connector-manager's /sdk/extract-text endpoint, without persisting.
# Returns the extracted text.
text = await ctx.content_storage.extract_text(
    data: bytes,
    mime_type: str,
    filename: str,
) -> str

# Same as extract_text, but also stores the extracted text and returns
# the resulting content_id.
content_id = await ctx.content_storage.extract_and_store_content(
    data: bytes,
    mime_type: str,
    filename: str,
) -> str

The returned content_id is a ULID that references the stored content.

Always prefer extract_text / extract_and_store_content over shelling out to your own PDF/Office parser. The connector-manager routes these calls through the centralized Docling service when the admin has enabled it (with the configured quality preset), and falls back to a lightweight built-in extractor otherwise — so your connector automatically honors the instance-wide document-conversion setting.

Document Model

The Document class represents a searchable document.

from omni_connector import Document, DocumentMetadata, DocumentPermissions

document = Document(
    external_id="unique-id-from-source",
    title="Document Title",
    content_id=content_id,  # From content_storage.save()
    metadata=DocumentMetadata(
        title="Document Title",
        author="Author Name",
        created_at=datetime.now(),
        updated_at=datetime.now(),
        mime_type="text/plain",
        size=1234,
        url="https://source.com/doc/123",
        path="/folder/document.txt",
        extra={"custom_field": "value"},
    ),
    permissions=DocumentPermissions(
        public=True,
        users=["user@example.com"],
        groups=["engineering"],
    ),
    attributes={"category": "documentation"},
)

Document Fields:

Field	Type	Required	Description
`external_id`	`str`	Yes	Unique ID from the source system
`title`	`str`	Yes	Document title for display
`content_id`	`str`	Yes	Reference to stored content
`metadata`	`DocumentMetadata`	No	Document metadata
`permissions`	`DocumentPermissions`	No	Access control
`attributes`	`dict`	No	Custom searchable attributes

Actions

Connectors can define custom actions that users can trigger from the Omni UI.

from omni_connector import ActionDefinition, ActionParameter

class MyConnector(Connector):
    @property
    def actions(self) -> list[ActionDefinition]:
        return [
            ActionDefinition(
                name="validate_connection",
                description="Test the API connection",
                parameters=[
                    ActionParameter(
                        name="timeout",
                        type="integer",
                        required=False,
                        description="Connection timeout in seconds",
                    ),
                ],
            ),
        ]

    async def execute_action(
        self,
        action: str,
        params: dict,
        credentials: dict
    ) -> dict:
        if action == "validate_connection":
            # Perform validation
            return {"status": "success", "message": "Connection valid"}
        raise ValueError(f"Unknown action: {action}")

Connector Skills

Connectors can ship skills — Markdown documents that guide the Omni agent through connector-specific workflows. Skills are loaded into Omni when the connector is enabled.

from omni_connector import ConnectorSkillDefinition
from pathlib import Path

class MyConnector(Connector):
    @property
    def skills(self) -> list[ConnectorSkillDefinition]:
        return [
            ConnectorSkillDefinition(
                id="my-tool-guide",
                title="My Tool Guide",
                description="Best practices for using My Tool",
                source_types=["my_source_type"],
                content=Path(__file__).parent.joinpath(
                    "skills/my-tool.md"
                ).read_text(),
            ),
        ]

Field	Type	Required	Description
`id`	`str`	Yes	Unique skill identifier
`title`	`str`	Yes	Human-readable skill name
`description`	`str`	No	Short summary shown in the skill list
`source_types`	`list[str]`	No	Source types the skill applies to
`content`	`str`	No	Markdown content loaded into the assistant
`mcp_prompt`	`str`	No	Name of an MCP prompt to use as the skill content instead of inline Markdown

MCP OAuth and Catalog Cache

Dynamic Client Registration (DCR)

For connectors whose MCP servers require per-user OAuth, OAuthManifestConfig supports automatic client registration so admins don’t need to create OAuth apps manually:

def oauth_config(self) -> OAuthManifestConfig | None:
    return OAuthManifestConfig(
        auth_endpoint="https://api.example.com/oauth/authorize",
        token_endpoint="https://api.example.com/oauth/token",
        registration_endpoint="https://api.example.com/oauth/register",
        token_endpoint_auth_method="none",  # Public DCR client
        resource="https://mcp.example.com",  # RFC 8707 resource indicator
        scope_sets={
            "read": OAuthScopeSet(
                read=["tasks:read"],
                write=["tasks:write"],
            ),
        },
    )

Field	Description
`registration_endpoint`	OAuth Dynamic Client Registration endpoint — Omni auto-creates a client instead of requiring admin setup
`token_endpoint_auth_method`	`"client_secret_post"`, `"client_secret_basic"`, or `"none"` (public DCR clients)
`resource`	RFC 8707 resource indicator, sent on auth/token requests (e.g., a remote MCP server URL)

Credential-Ready Lifecycle

After a user completes OAuth, connector-manager calls oauth_credential_ready so the connector can run post-OAuth setup such as authenticated MCP catalog discovery:

from omni_connector import OAuthCredentialReadyRequest

class MyConnector(Connector):
    async def oauth_credential_ready(
        self, request: OAuthCredentialReadyRequest
    ) -> None:
        # Discover MCP tools now that credentials are available
        self._load_mcp_catalog_cache(self.mcp_adapter)

Catalog Cache

The SDK caches MCP tool catalogs to disk (default: /var/lib/omni/mcp-catalogs) to avoid costly discovery on every restart. Set CATALOG_CACHE_TTL_SECONDS to control the cache lifetime; it defaults to 86400 seconds (24 hours).

CATALOG_CACHE_TTL_SECONDS=3600

Set a shorter TTL for MCP servers whose tool sets change frequently, or a larger value for stable tool catalogs. Here’s a complete example of an RSS feed connector:

import asyncio
from datetime import datetime
from typing import Any

import feedparser
from omni_connector import (
    ActionDefinition,
    ActionParameter,
    Connector,
    Document,
    DocumentMetadata,
    DocumentPermissions,
    SyncContext,
)


class RSSConnector(Connector):
    """Connector that indexes RSS/Atom feeds."""

    @property
    def name(self) -> str:
        return "rss"

    @property
    def version(self) -> str:
        return "1.0.0"

    @property
    def sync_modes(self) -> list[str]:
        return ["full", "incremental"]

    @property
    def actions(self) -> list[ActionDefinition]:
        return [
            ActionDefinition(
                name="validate_feed",
                description="Validate that a feed URL is accessible",
                parameters=[
                    ActionParameter(
                        name="url",
                        type="string",
                        required=True,
                        description="The RSS/Atom feed URL to validate",
                    ),
                ],
            ),
        ]

    async def sync(
        self,
        source_config: dict[str, Any],
        credentials: dict[str, Any],
        checkpoint: dict[str, Any] | None,
        ctx: SyncContext,
    ) -> None:
        """Sync articles from an RSS feed."""
        feed_url = source_config.get("feed_url")
        if not feed_url:
            await ctx.fail("Missing feed_url in source configuration")
            return

        # Parse the feed
        feed = await asyncio.to_thread(feedparser.parse, feed_url)

        if feed.bozo and not feed.entries:
            await ctx.fail(f"Failed to parse feed: {feed.bozo_exception}")
            return

        # Get last sync time for incremental sync
        last_sync_time = None
        if checkpoint and "last_sync_time" in checkpoint:
            last_sync_time = datetime.fromisoformat(checkpoint["last_sync_time"])

        for entry in feed.entries:
            if ctx.is_cancelled():
                return

            await ctx.increment_scanned()

            # Skip old entries in incremental mode
            if last_sync_time and hasattr(entry, "published_parsed"):
                entry_date = datetime(*entry.published_parsed[:6])
                if entry_date <= last_sync_time:
                    continue

            # Build content from entry
            content = self._build_content(entry)
            content_id = await ctx.content_storage.save(content)

            # Create document
            doc = Document(
                external_id=entry.get("id", entry.link),
                title=entry.get("title", "Untitled"),
                content_id=content_id,
                metadata=DocumentMetadata(
                    author=entry.get("author"),
                    url=entry.get("link"),
                    created_at=self._parse_date(entry),
                    extra={
                        "feed_title": feed.feed.get("title"),
                        "feed_url": feed_url,
                    },
                ),
                permissions=DocumentPermissions(public=True),
            )

            await ctx.emit(doc)

        # Complete with updated checkpoint
        await ctx.complete({"last_sync_time": datetime.utcnow().isoformat()})

    def _build_content(self, entry) -> str:
        """Extract content from feed entry."""
        if hasattr(entry, "content") and entry.content:
            return entry.content[0].get("value", "")
        if hasattr(entry, "summary"):
            return entry.summary
        return entry.get("title", "")

    def _parse_date(self, entry) -> datetime | None:
        """Parse date from feed entry."""
        if hasattr(entry, "published_parsed") and entry.published_parsed:
            return datetime(*entry.published_parsed[:6])
        return None

    async def execute_action(
        self,
        action: str,
        params: dict[str, Any],
        credentials: dict[str, Any],
    ) -> dict[str, Any]:
        if action == "validate_feed":
            url = params.get("url")
            if not url:
                return {"status": "error", "error": "Missing URL parameter"}

            feed = await asyncio.to_thread(feedparser.parse, url)
            if feed.bozo and not feed.entries:
                return {"status": "error", "error": str(feed.bozo_exception)}

            return {
                "status": "success",
                "result": {
                    "title": feed.feed.get("title"),
                    "entry_count": len(feed.entries),
                },
            }

        return {"status": "error", "error": f"Unknown action: {action}"}


if __name__ == "__main__":
    RSSConnector().serve(port=8000)

Error Handling

The SDK provides custom exception classes:

from omni_connector import (
    ConnectorError,      # Base exception
    SdkClientError,      # Communication errors
    SyncCancelledError,  # Sync was cancelled
    ConfigurationError,  # Configuration issues
)

Handling errors in sync:

async def sync(self, source_config, credentials, checkpoint, ctx):
    try:
        # Your sync logic
        await ctx.complete()
    except SyncCancelledError:
        # Handle cancellation gracefully
        return
    except Exception as e:
        await ctx.fail(str(e))

Development

Project Setup

# From your connector directory inside the omni monorepo
uv sync

Running Tests

uv run pytest tests/ -v

Type Checking

uv run mypy my_connector/

Linting

uv run ruff check my_connector/

Testing

The SDK includes a testing module (omni_connector.testing) with utilities for writing integration tests for your connectors.

TestHarness

The TestHarness class provides a self-contained environment for testing your connector without needing a running Omni deployment.

import pytest
from omni_connector.testing import TestHarness
from my_connector import MyConnector

@pytest.fixture
def harness():
    return TestHarness(MyConnector())

@pytest.mark.asyncio
async def test_full_sync(harness):
    # Seed test data
    harness.set_source_config({"api_url": "https://api.example.com"})
    harness.set_credentials({"api_key": "test-key"})

    # Run sync
    result = await harness.run_sync()

    # Assert results
    assert result.status == "completed"
    assert result.documents_emitted > 0

@pytest.mark.asyncio
async def test_incremental_sync(harness):
    # Run initial sync
    harness.set_source_config({"api_url": "https://api.example.com"})
    harness.set_credentials({"api_key": "test-key"})
    first_result = await harness.run_sync()

    # Run incremental sync with checkpoint from first run
    second_result = await harness.run_sync(checkpoint=first_result.checkpoint)
    assert second_result.status == "completed"

Assertions

The harness captures all emitted documents, errors, and checkpoint changes for easy assertion:

async def test_document_content(harness):
    harness.set_source_config({"feed_url": "https://example.com/feed"})
    result = await harness.run_sync()

    # Check emitted documents
    assert len(result.documents) == 5
    assert result.documents[0].title == "Expected Title"

    # Check for errors
    assert len(result.errors) == 0

    # Check final checkpoint
    assert "last_sync_time" in result.checkpoint

API Reference

Exports

The SDK exports the following: Core Classes:

Connector - Abstract base class
SyncContext - Sync operation context
ContentStorage - Content storage interface
SdkClient - SDK client for connector-manager

Data Models:

Document, DocumentMetadata, DocumentPermissions
ConnectorEvent, EventType
ActionDefinition, ActionParameter
ActionRequest, ActionResponse
ConnectorManifest, SyncMode
SyncRequest, SyncResponse
CancelRequest, CancelResponse
ConnectorSkillDefinition
OAuthManifestConfig, OAuthScopeSet
OAuthCredentialReadyRequest
McpResourceDefinition, McpPromptDefinition
StdioMcpServer, HttpMcpServer

Exceptions:

ConnectorError, SdkClientError, SyncCancelledError, ConfigurationError

What’s Next

SDK Overview

Learn about SDK architecture

TypeScript SDK

Build connectors with TypeScript

Getting Started

Deployment

Connectors

Developers

User Guide

Administration

Installation

Quick Start

Core Concepts

Connector Class

SyncContext

ContentStorage

Document Model

Actions

Connector Skills

MCP OAuth and Catalog Cache

Dynamic Client Registration (DCR)

Credential-Ready Lifecycle

Catalog Cache

Error Handling

Development

Project Setup

Running Tests

Type Checking

Linting

Testing

TestHarness

Assertions

API Reference

Exports

What’s Next

SDK Overview

TypeScript SDK

​Installation

​Quick Start

​Core Concepts

​Connector Class

​SyncContext

​ContentStorage

​Document Model

​Actions

​Connector Skills

​MCP OAuth and Catalog Cache

​Dynamic Client Registration (DCR)

​Credential-Ready Lifecycle

​Catalog Cache

​Error Handling

​Development

​Project Setup

​Running Tests

​Type Checking

​Linting

​Testing

​TestHarness

​Assertions

​API Reference

​Exports

​What’s Next

SDK Overview

TypeScript SDK

Installation

Quick Start

Core Concepts

Connector Class

SyncContext

ContentStorage

Document Model

Actions

Connector Skills

MCP OAuth and Catalog Cache

Dynamic Client Registration (DCR)

Credential-Ready Lifecycle

Catalog Cache

Error Handling

Development

Project Setup

Running Tests

Type Checking

Linting

Testing

TestHarness

Assertions

API Reference

Exports

What’s Next