Skip to content

Integrating groundworkers

How to drive Groundworkers from Python against the same steady-state configuration the setup console produces. Nothing here imports Textual, Groundskeeping, or another application's setup code.

Build the runtime

from groundworkers.app import build_application
from groundworkers.bootstrap import build_app_config

config = build_app_config(config_path="/path/to/config.toml")
app = build_application(config)

build_app_config resolves the stack and returns an AppConfig holding only Groundworkers-owned values:

Attribute Type
groundworkers GroundworkersConfig
cdm_database ResolvedCDMDatabase
cdm_engine, cdm_vocab_engine sqlalchemy.Engine
cdm_session_factory Routed SQLAlchemy session factory for primary, vocabulary, and results roles
embedding_model ResolvedModel \| None
vector_store ResolvedVectorStore \| None

Omitting config_path uses OA_CONFIG_PATH or the default stack location.

Available services

app.services attributes are None only when their prerequisites are absent, so check before use:

services = app.services

services.vocab          # always, with a CDM database
services.graph          # always, with a CDM database
services.grounding      # always, with a CDM database
services.mapping        # always, with a CDM database
services.text           # only when groundworkers.llm_model_name resolves
services.domain         # only when groundworkers.llm_model_name resolves
services.source_planning

A CDM-only configuration is fully supported. Graph availability follows the resolved CDM database — there is no separate section to enable.

The engines are created by omop_alchemy.config.create_cdm_engines(..., register_claims=False), and cdm_session_factory comes from omop_alchemy.cross_database.cdm_sessionmaker. Use the session factory for queries that may touch more than one OMOP role. On PostgreSQL, engine creation checks role schema claims eagerly, so the configured databases must be reachable while build_app_config() runs.

Grounding free text

result = app.services.grounding.ground(
    "type 2 diabetes",
    limit=10,
    domain="Condition",
    vocabulary_id=None,
    parent_ids=None,        # optional required-ancestor constraint
    standard_only=False,    # narrows candidate resolution, not results
    active_only=False,
)

for hit in result["results"]:
    if hit["standard_concept"]:
        print(hit["concept_id"], hit["concept_name"])
    elif hit["classification_concept"]:
        # A real hierarchy node, but not a valid CDM entity-field target.
        print("classification:", hit["concept_id"], hit["concept_name"])

limit must be positive; a non-positive value raises GroundworkersError with code INVALID_INPUT.

Interpreting the explanation

explanation = result["grounding_explanation"]
explanation["matched_tier"]           # EXACT | FULLTEXT | EMBEDDING_NEAREST | PARTIAL
explanation["used_embedding"]         # bool
explanation["embedding_tier_detail"]  # None, or why the embedding tier was skipped

Tiers run in order and the first one that yields results wins. When embedding_tier_detail is set, the embedding tier was planned but could not run — usually an unreachable provider — and the answer came from lexical tiers alone. Treat a set value as "this result may be incomplete".

Error handling

Every service raises GroundworkersError with a stable code:

from groundworkers.base.errors import GroundworkersError

try:
    concept = app.services.graph.get_concept(201826)
except GroundworkersError as exc:
    print(exc.code, exc.message)   # NOT_FOUND | INVALID_INPUT | BACKEND_UNAVAIL | QUERY_ERROR | …

Messages are safe to log and surface: provider and database failures are translated so they do not carry credentials or connection strings.

Concept flag contract

Field Meaning
standard_concept true only for raw standard_concept = 'S'
classification_concept true only for raw standard_concept = 'C'
is_active invalid_reason unset, treating blank and whitespace-only as active

Do not derive standardness from omop-graph's ConceptView.standard_concept: that field is a single boolean covering both 'S' and 'C'. Groundworkers reads the raw flag at its adapter boundary so callers get the distinction.

Writing configuration

Configuration writes are a local console concern, not part of the integration surface. If you are embedding the setup console in another host application, drive GroundworkersConfigMutationService through Groundskeeping's ConfigWizardController:

from groundskeeping.configurator import ConfigWizardController, MutationOperation

from groundworkers.application.setup.configuration_provider import (
    CDM_SETUP_TARGET,
    GroundworkersConfigMutationService,
    cdm_setup_workflow,
)
from groundworkers.application.setup.model_inventory import discover_provider_models

service = GroundworkersConfigMutationService(
    "/path/to/config.toml",
    model_discoverer=discover_provider_models,
)
operation = (
    MutationOperation.UPDATE
    if service.capabilities(CDM_SETUP_TARGET, MutationOperation.UPDATE).supported
    else MutationOperation.CREATE
)
controller = ConfigWizardController(cdm_setup_workflow(operation), service)

The same shape covers MODEL_SETUP_TARGET / model_setup_workflow, LLM_SETUP_TARGET / llm_setup_workflow, and VECTOR_STORE_SETUP_TARGET / vector_store_setup_workflow. The provider decides create-versus-update from the current file, owns revision checking and redaction, and returns conflicted rather than clobbering another writer. model_discoverer is the injected seam for live model inventory; supply your own to avoid network calls in tests.

This requires the tui extra for Groundskeeping, but not Textual — the provider and controller are host-agnostic.