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.
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-sdkCodex
codex plugin marketplace add gleanwork/glean-indexing-sdk
codex plugin add glean-connector-builder@glean-indexing-sdkCursor 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.
- Python >= 3.10
- A Glean instance and an indexing API token
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 MonitoringEvery 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")| Capability | What it gives you |
|---|---|
| Connector types | Four base classes: in-memory, sync streaming, async streaming, and people/identity. |
| Pull integrations | PullHttpClient with retries and backoff, link/offset/cursor pagination, and token-bucket rate limiting. |
| Push & indexing | PushUploader for documents, users, groups, memberships, and employees, with parallel batch uploads. |
| Permissions | Per-document ACLs and datasource identities so results respect who can see what. |
| Testing | Three phases: fully mocked, real-source-with-record/replay, and live end-to-end. |
| Observability | Structured logging and metrics, with optional CloudWatch and Google Cloud plugins. |
| Status & debugging | StatusClient and glean-idx document status to answer "why isn't my document in search?" |
| Deployment | glean-idx deploy generates Docker and Terraform for AWS or GCP. |
| Connector Builder | An agent plugin that builds a connector from a description of your source. |
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 CronJobCommands 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 doctorrun, 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 runEvery 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.documentsglean-idx schema document prints the JSON Schema your transform() has to
produce, and glean-idx completion zsh sets up tab completion.
connector.index_data(mode=IndexingMode.FULL) # re-index everything
connector.index_data(mode=IndexingMode.INCREMENTAL) # only changes since the last crawlA 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.
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 formatArchitecture notes for contributors live in docs/.