Glean Indexing SDK

Build custom Glean connectors in Python. The SDK handles fetching, transforming, batching, and uploading your data to Glean's indexing APIs, so you write only the parts that are specific to your source.

📖 Full documentation on the Glean Developer site.

Build one with an agent

The fastest path is to let a coding agent build the connector. Your agent installs it straight from this repository, which doubles as the plugin marketplace.

Claude Code

claude plugin marketplace add gleanwork/glean-indexing-sdk
claude plugin install glean-connector-builder@glean-indexing-sdk

Codex

codex plugin marketplace add gleanwork/glean-indexing-sdk
codex plugin add glean-connector-builder@glean-indexing-sdk

Cursor has no plugin CLI — open Dashboard → Plugins → Add Marketplace → Import from Repo, point it at gleanwork/glean-indexing-sdk, then install from Customize.

Then describe your source:

I want to push my Webex data to Glean. Build a connector for me.

The agent explores the source's API, confirms a plan with you, generates the connector against this SDK, and tests it. See the Indexing SDK overview for what it does and what to review in the output.

Or write it yourself

Requirements

• Python >= 3.10
• A Glean instance and an indexing API token

Installation

pip install glean-indexing-sdk

# Optional cloud observability plugins
pip install "glean-indexing-sdk[aws]"   # CloudWatch logs + metrics
pip install "glean-indexing-sdk[gcp]"   # Cloud Logging + Cloud Monitoring

Quickstart

Every connector has two parts: a data client that fetches from your source, and a connector that transforms the result into Glean documents. The flow is fetch → transform → upload; you implement get_source_data() and transform(), and the SDK does the rest.

Set your credentials:

export GLEAN_SERVER_URL="https://your-company-be.glean.com"
export GLEAN_INDEXING_API_TOKEN="your-indexing-api-token"

Then define and run a connector:

from datetime import datetime
from typing import Any, List, Optional, Sequence, TypedDict

from glean.indexing.connectors import BaseDataClient, BaseDatasourceConnector
from glean.indexing.models import (
    ContentDefinition,
    CustomDatasourceConfig,
    DocumentDefinition,
    IndexingMode,
    UserReferenceDefinition,
)


class WikiPage(TypedDict):
    id: str
    title: str
    content: str
    author: str
    updated_at: str
    url: str


class WikiDataClient(BaseDataClient[WikiPage]):
    """Fetches pages from the source system. Replace the body with a real API call."""

    def __init__(self, base_url: str, api_token: str):
        self.base_url = base_url
        self.api_token = api_token

    def get_source_data(self, since: Optional[str] = None, **kwargs: Any) -> Sequence[WikiPage]:
        return [
            {
                "id": "page_123",
                "title": "Engineering Onboarding Guide",
                "content": "Welcome to the engineering team...",
                "author": "jane.smith@company.com",
                "updated_at": "2026-02-01T14:30:00Z",
                "url": f"{self.base_url}/pages/123",
            }
        ]


class CompanyWikiConnector(BaseDatasourceConnector[WikiPage]):
    """Transforms wiki pages into Glean documents."""

    configuration = CustomDatasourceConfig(
        name="company_wiki",
        display_name="Company Wiki",
        url_regex=r"https://wiki\.company\.com/.*",
        is_user_referenced_by_email=True,
    )

    def transform(self, data: Sequence[WikiPage]) -> List[DocumentDefinition]:
        return [
            DocumentDefinition(
                id=page["id"],
                title=page["title"],
                datasource=self.name,
                view_url=page["url"],
                body=ContentDefinition(mime_type="text/plain", text_content=page["content"]),
                author=UserReferenceDefinition(email=page["author"]),
                # created_at / updated_at are epoch seconds, not ISO strings.
                updated_at=int(
                    datetime.fromisoformat(page["updated_at"].replace("Z", "+00:00")).timestamp()
                ),
            )
            for page in data
        ]


if __name__ == "__main__":
    connector = CompanyWikiConnector(
        name="company_wiki",
        data_client=WikiDataClient(
            base_url="https://wiki.company.com", api_token="your-wiki-token"
        ),
    )
    connector.configure_datasource()
    connector.index_data(mode=IndexingMode.FULL)

Test it without touching the network:

from glean.indexing.testing import StaticDataClient, run_connector

result = run_connector(CompanyWikiConnector("company_wiki", StaticDataClient([...])))
result.assert_documents_posted(count=1, datasource="company_wiki")

What's in the box

The CLI

One command, glean-idx, covers the whole loop.

glean-idx doctor                    # are my credentials right?
glean-idx validate ./my-connector   # is the plan complete, before writing code?
glean-idx test --phase all          # mocked, then real source, then live
glean-idx run                       # crawl for real
glean-idx datasource status --datasource my-source
glean-idx document status --datasource my-source --document Article doc-1
glean-idx deploy init --cloud gcp   # Docker and Terraform for a CronJob

Commands split into two kinds, and glean-idx --help says which is which.

Most need only GLEAN_SERVER_URL and GLEAN_INDEXING_API_TOKEN, so they run anywhere, including with no install at all:

uvx --from glean-indexing-sdk glean-idx doctor

run, test, and datasource configure import your connector, so they run inside the connector project with the SDK installed alongside your code:

uv run glean-idx run

Every command takes --output json for a stable envelope, --yes to skip confirmations unattended, and returns a documented exit code — 3 for a missing precondition, 4 for a Glean error, 5 for a validation failure. In JSON mode the envelope goes to stdout whether it succeeded or not, so there is one stream to read:

glean-idx datasource status --datasource my-source --output json | jq .data.documents

glean-idx schema document prints the JSON Schema your transform() has to produce, and glean-idx completion zsh sets up tab completion.

Indexing modes

connector.index_data(mode=IndexingMode.FULL)         # re-index everything
connector.index_data(mode=IndexingMode.INCREMENTAL)  # only changes since the last crawl

A full crawl replaces the indexed state: documents absent from the run are deleted as stale. Incremental passes a since timestamp to your data client, but the SDK does not persist checkpoints — override _get_last_crawl_timestamp() on your connector to supply one. See Indexing modes.

Contributing

This project uses mise for toolchain management and uv for Python dependencies. See CONTRIBUTING.md.

mise run setup    # create venv and install dependencies
mise run test     # run all tests
mise run lint     # ruff, pyright, markdown-code
mise run lint:fix # auto-fix and format

Architecture notes for contributors live in docs/.

License

MIT