Connector SDK Overview

/build-connector Asana

┌─────────────────┐         ┌───────────────────┐         ┌─────────────┐
│  Your Connector │ ←HTTP→  │ connector-manager │ ←Queue→ │ omni-indexer│
│   (SDK-based)   │         │   (orchestrator)  │         │  (indexing) │
└─────────────────┘         └───────────────────┘         └─────────────┘
        │                            │
        │                            │
        ↓                            ↓
┌─────────────────┐         ┌───────────────────┐
│   Data Source   │         │    PostgreSQL     │
│   (your API)    │         │   (storage/queue) │
└─────────────────┘         └───────────────────┘

from omni_connector import Connector, Document, DocumentMetadata, 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"]

    async def sync(
        self,
        source_config: dict,
        credentials: dict,
        checkpoint: dict | None,
        ctx: SyncContext
    ) -> None:
        # Fetch items from your data source
        items = await self.fetch_items(credentials["api_key"])

        for item in items:
            # Store content
            content_id = await ctx.content_storage.save(item["content"])

            # Emit document
            await ctx.emit(Document(
                external_id=item["id"],
                title=item["title"],
                content_id=content_id,
                metadata=DocumentMetadata(
                    url=item["url"],
                    created_at=item["created_at"],
                ),
            ))

            await ctx.increment_scanned()

        await ctx.complete()

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

from omni_connector import Connector, StdioMcpServer

class MyConnector(Connector):
    @property
    def mcp_server(self) -> StdioMcpServer:
        return StdioMcpServer(
            command="my-mcp-server",
            args=["stdio"],
        )

    def prepare_mcp_env(self, credentials: dict) -> dict[str, str]:
        return {"API_TOKEN": credentials["api_key"]}

from omni_connector import Connector, HttpMcpServer

class MyConnector(Connector):
    @property
    def mcp_server(self) -> HttpMcpServer:
        return HttpMcpServer(url="https://api.example.com/mcp")

    def prepare_mcp_headers(self, credentials: dict) -> dict[str, str]:
        return {"Authorization": f"Bearer {credentials['api_key']}"}

from omni_connector import ConnectorSkillDefinition

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(),
            ),
        ]

class MyConnector(Connector):
    async def oauth_credential_ready(
        self, request: OAuthCredentialReadyRequest
    ) -> None:
        # Discover the MCP tool catalog now that credentials are available
        pass

services:
  my-connector:
    build:
      context: ../connectors/my-connector
    profiles: ["my-connector"]
    environment:
      CONNECTOR_MANAGER_URL: ${CONNECTOR_MANAGER_URL}
      CONNECTOR_HOST_NAME: my-connector
      PORT: ${MY_CONNECTOR_PORT}
    networks:
      - omni-network