Platform

Shared infrastructure: config, types, errors, security, and logging. Used throughout the application. No business logic.

Overview

• Config — Application settings (API keys, database URL, feature flags). Loaded from environment and .env files.

• Types — Shared type aliases: JSONObject, JSONArray, JSONValue.

• Errors — WDKError, ValidationError, ErrorCode. Consistent error handling.

• Security — Auth and authorization helpers.

• Logging — Structured logging setup.

Design Decisions

Redis for event bus

The platform event bus uses Redis Streams for durable event delivery. Chat streams and experiment executions publish events that clients consume via SSE. Redis Streams support consumer groups, cursor-based replay, and TTL-based cleanup — all necessary for reliable long-running operations that may outlive HTTP connections.

Pydantic settings

Configuration uses pydantic-settings with TOML + environment variable layering. TOML provides checked-in defaults; environment variables override for deployment. This avoids the “which .env file?” problem while keeping sensitive values out of version control.

Structured logging via structlog

All logging uses structlog with JSON output. This enables structured queries in log aggregation tools (filtering by user_id, strategy_id, tool_name) without string parsing. Development mode uses human-readable console output.

Context variables for request state

Request-scoped state (auth token, user ID, site context) propagates via Python contextvars. This avoids threading state through every function signature while remaining async-safe (each task gets its own context copy).

Tip

All configuration values can be overridden via environment variables. See config.toml for defaults and platform/config.py for the full settings schema.

Config

Purpose: Application settings. API keys (OpenAI, Anthropic, etc.), database URL, Qdrant URL, feature flags (rag_enabled). Loaded via pydantic-settings.

Key function: get_settings()

Application configuration using pydantic-settings.

class veupath_chatbot.platform.config.TomlConfigSettingsSource(settings_cls)

Bases: PydanticBaseSettingsSource

Load settings from a TOML config file.

__init__(settings_cls)

get_field_value(field, field_name)

Gets the value, the key for model creation, and a flag to determine whether value is complex.

This is an abstract method that should be overridden in every settings source classes.

Args:

field: The field. field_name: The field name.

Returns:

A tuple that contains the value, key and a flag to determine whether value is complex.

Return type:

tuple[object, str, bool]

class veupath_chatbot.platform.config.Settings(_case_sensitive=None, _nested_model_default_partial_update=None, _env_prefix=None, _env_prefix_target=None, _env_file=PosixPath('.'), _env_file_encoding=None, _env_ignore_empty=None, _env_nested_delimiter=None, _env_nested_max_split=None, _env_parse_none_str=None, _env_parse_enums=None, _cli_prog_name=None, _cli_parse_args=None, _cli_settings_source=None, _cli_parse_none_str=None, _cli_hide_none_type=None, _cli_avoid_json=None, _cli_enforce_required=None, _cli_use_class_docs_for_groups=None, _cli_exit_on_error=None, _cli_prefix=None, _cli_flag_prefix_char=None, _cli_implicit_flags=None, _cli_ignore_unknown_args=None, _cli_kebab_case=None, _cli_shortcuts=None, _secrets_dir=None, _build_sources=None, *, api_host='0.0.0.0', api_port=8000, api_env='development', api_debug=False, api_secret_key='dev-only-secret-key-change-in-prod', api_docs_enabled=True, database_url='postgresql+asyncpg://postgres:postgres@localhost:5432/pathfinder', redis_url='redis://localhost:6379/0', openai_api_key='', openai_model='gpt-4.1', openai_temperature=0.0, openai_top_p=1.0, openai_hyperparams=<factory>, anthropic_api_key='', anthropic_model='claude-sonnet-4-6', anthropic_temperature=0.0, anthropic_top_p=1.0, anthropic_hyperparams=<factory>, gemini_api_key='', gemini_model='gemini-2.5-pro', gemini_temperature=0.0, gemini_top_p=1.0, gemini_hyperparams=<factory>, ollama_base_url='http://localhost:11434/v1', rag_enabled=True, qdrant_url='http://localhost:6333', qdrant_api_key=None, qdrant_timeout_seconds=10.0, rag_startup_max_strategies_per_site=None, rag_startup_public_strategies_concurrency=None, rag_startup_public_strategies_llm_model='gpt-4.1-nano', rag_startup_public_strategies_report_path='/tmp/ingest_public_strategies_report.jsonl', embeddings_model='text-embedding-3-small', embeddings_base_url='', subkani_model='gpt-4.1-mini', subkani_temperature=0.0, subkani_top_p=1.0, subkani_max_concurrency=6, subkani_timeout_seconds=120, default_model_id='openai/gpt-4.1', default_reasoning_effort='medium', veupathdb_default_site='veupathdb', veupathdb_sites_config=None, veupathdb_cache_ttl=3600, veupathdb_auth_token=None, veupathdb_oauth_url=None, veupathdb_oauth_client_id=None, PATHFINDER_CHAT_PROVIDER='default', log_level='INFO', log_format='json', cors_origins=['http://localhost:3000'], cors_origin_regex='^https?://(localhost|127\\.0\\.0\\.1)(:\\d+)?$')

Bases: BaseSettings

Application settings loaded from environment variables.

model_config = {'arbitrary_types_allowed': True, 'case_sensitive': False, 'cli_avoid_json': False, 'cli_enforce_required': False, 'cli_exit_on_error': True, 'cli_flag_prefix_char': '-', 'cli_hide_none_type': False, 'cli_ignore_unknown_args': False, 'cli_implicit_flags': False, 'cli_kebab_case': False, 'cli_parse_args': None, 'cli_parse_none_str': None, 'cli_prefix': '', 'cli_prog_name': None, 'cli_shortcuts': None, 'cli_use_class_docs_for_groups': False, 'enable_decoding': True, 'env_file': '/home/docs/checkouts/readthedocs.org/user_builds/veupathdb-pathfinder/checkouts/latest/.env', 'env_file_encoding': 'utf-8', 'env_ignore_empty': False, 'env_nested_delimiter': None, 'env_nested_max_split': None, 'env_parse_enums': None, 'env_parse_none_str': None, 'env_prefix': '', 'env_prefix_target': 'variable', 'extra': 'ignore', 'json_file': None, 'json_file_encoding': None, 'nested_model_default_partial_update': False, 'protected_namespaces': ('model_validate', 'model_dump', 'settings_customise_sources'), 'secrets_dir': None, 'toml_file': None, 'validate_default': True, 'yaml_config_section': None, 'yaml_file': None, 'yaml_file_encoding': None}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

api_host: str

api_port: int

api_env: Literal['development', 'staging', 'production']

api_debug: bool

api_secret_key: str

api_docs_enabled: bool

database_url: str

redis_url: str

openai_api_key: str

openai_model: str

openai_temperature: float

openai_top_p: float

openai_hyperparams: dict[str, object]

anthropic_api_key: str

anthropic_model: str

anthropic_temperature: float

anthropic_top_p: float

anthropic_hyperparams: dict[str, object]

gemini_api_key: str

gemini_model: str

gemini_temperature: float

gemini_top_p: float

gemini_hyperparams: dict[str, object]

ollama_base_url: str

rag_enabled: bool

qdrant_url: str

qdrant_api_key: str | None

qdrant_timeout_seconds: float

rag_startup_max_strategies_per_site: int | None

rag_startup_public_strategies_concurrency: int | None

rag_startup_public_strategies_llm_model: str

rag_startup_public_strategies_report_path: str

embeddings_model: str

embeddings_base_url: str

subkani_model: str

subkani_temperature: float

subkani_top_p: float

subkani_max_concurrency: int

subkani_timeout_seconds: int

default_model_id: str

default_reasoning_effort: Literal['none', 'low', 'medium', 'high']

veupathdb_default_site: str

veupathdb_sites_config: str | None

veupathdb_cache_ttl: int

veupathdb_auth_token: str | None

veupathdb_oauth_url: str | None

veupathdb_oauth_client_id: str | None

chat_provider: str

log_level: str

log_format: Literal['json', 'console']

cors_origins: list[str]

cors_origin_regex: str | None

property is_development: bool

Check if running in development mode.

property is_production: bool

Check if running in production mode.

model_post_init(_Settings__context)

Validate settings after initialization.

classmethod settings_customise_sources(settings_cls, init_settings, env_settings, dotenv_settings, file_secret_settings)

Define the sources and their order for loading the settings values.

Args:

settings_cls: The Settings class. init_settings: The InitSettingsSource instance. env_settings: The EnvSettingsSource instance. dotenv_settings: The DotEnvSettingsSource instance. file_secret_settings: The SecretsSettingsSource instance.

Returns:

A tuple containing the sources and their order for loading the settings values.

Return type:

tuple[PydanticBaseSettingsSource, …]

veupath_chatbot.platform.config.get_settings()

Get cached settings instance.

Return type:

Settings

Types

Purpose: Shared type aliases for untyped dict/list payloads. JSONObject, JSONArray, JSONValue. Used throughout the codebase.

Common type aliases for the codebase.

type veupath_chatbot.platform.types.JSONValue = JsonValue

Type alias for JSON values.

type veupath_chatbot.platform.types.JSONObject = dict[str, JSONValue]

Type alias for JSON objects (dictionaries with string keys).

type veupath_chatbot.platform.types.JSONArray = list[JSONValue]

Type alias for JSON arrays.

veupath_chatbot.platform.types.ModelProvider

Supported LLM provider identifiers.

alias of Literal[‘openai’, ‘anthropic’, ‘google’, ‘ollama’, ‘mock’]

veupath_chatbot.platform.types.ReasoningEffort

Reasoning effort level for models that support it.

alias of Literal[‘none’, ‘low’, ‘medium’, ‘high’]

veupath_chatbot.platform.types.as_json_object(value)

Type guard: assert that a JSONValue is a JSONObject.

Parameters:

value (JSONValue) – Value to check.

Returns:

Same value as JSONObject.

Raises:

TypeError – If value is not a dict.

Return type:

JSONObject

veupath_chatbot.platform.types.as_json_array(value)

Type guard: assert that a JSONValue is a JSONArray.

Parameters:

value (JSONValue) – Value to check.

Returns:

Same value as JSONArray.

Raises:

TypeError – If value is not a list.

Return type:

JSONArray

Errors

Purpose: Error codes and exception types. WDKError for WDK API failures, ValidationError for plan validation, ErrorCode enum. Used for consistent HTTP error responses.

Key classes: WDKError, ValidationError

Typed error model with problem+json responses.

class veupath_chatbot.platform.errors.ErrorCode(*values)

Bases: StrEnum

Application error codes.

INTERNAL_ERROR = 'INTERNAL_ERROR'

VALIDATION_ERROR = 'VALIDATION_ERROR'

NOT_FOUND = 'NOT_FOUND'

UNAUTHORIZED = 'UNAUTHORIZED'

FORBIDDEN = 'FORBIDDEN'

RATE_LIMITED = 'RATE_LIMITED'

SITE_NOT_FOUND = 'SITE_NOT_FOUND'

SEARCH_NOT_FOUND = 'SEARCH_NOT_FOUND'

INVALID_PARAMETERS = 'INVALID_PARAMETERS'

WDK_ERROR = 'WDK_ERROR'

STRATEGY_NOT_FOUND = 'STRATEGY_NOT_FOUND'

INVALID_STRATEGY = 'INVALID_STRATEGY'

STEP_NOT_FOUND = 'STEP_NOT_FOUND'

INCOMPATIBLE_STEPS = 'INCOMPATIBLE_STEPS'

ENSURE_SINGLE_OUTPUT_FAILED = 'ENSURE_SINGLE_OUTPUT_FAILED'

CONVERSATION_NOT_FOUND = 'CONVERSATION_NOT_FOUND'

class veupath_chatbot.platform.errors.ProblemDetail(*, type='about:blank', title, status, detail=None, instance=None, code, errors=None)

Bases: BaseModel

RFC 7807 Problem Details response.

type: str

title: str

status: int

detail: str | None

instance: str | None

code: ErrorCode

errors: JSONArray | None

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

exception veupath_chatbot.platform.errors.AppError(code, title, status=400, detail=None, errors=None)

Bases: Exception

Base application error.

__init__(code, title, status=400, detail=None, errors=None)

exception veupath_chatbot.platform.errors.InternalError(title='Internal error', detail=None)

Bases: AppError

Internal server error (unexpected invariant failure).

__init__(title='Internal error', detail=None)

exception veupath_chatbot.platform.errors.NotFoundError(code=ErrorCode.NOT_FOUND, title='Resource not found', detail=None)

Bases: AppError

Resource not found error.

__init__(code=ErrorCode.NOT_FOUND, title='Resource not found', detail=None)

exception veupath_chatbot.platform.errors.UnauthorizedError(code=ErrorCode.UNAUTHORIZED, title='Unauthorized', detail=None)

Bases: AppError

Unauthorized error.

__init__(code=ErrorCode.UNAUTHORIZED, title='Unauthorized', detail=None)

exception veupath_chatbot.platform.errors.ForbiddenError(code=ErrorCode.FORBIDDEN, title='Forbidden', detail=None)

Bases: AppError

Forbidden error.

__init__(code=ErrorCode.FORBIDDEN, title='Forbidden', detail=None)

exception veupath_chatbot.platform.errors.ValidationError(title='Validation failed', detail=None, errors=None)

Bases: AppError

Validation error.

__init__(title='Validation failed', detail=None, errors=None)

exception veupath_chatbot.platform.errors.WDKError(detail, status=502)

Bases: AppError

Error from VEuPathDB WDK service.

__init__(detail, status=502)

async veupath_chatbot.platform.errors.app_error_handler(request, exc)

Handle AppError exceptions.

Return type:

JSONResponse

async veupath_chatbot.platform.errors.http_exception_handler(request, exc)

Handle FastAPI HTTPException.

Return type:

JSONResponse

Security

Purpose: Authentication and authorization. Token validation, permission checks, user context. Used by HTTP deps and routers.

Authentication, authorization, and rate limiting.

async veupath_chatbot.platform.security.get_optional_user(request, cookie_token=None)

Get current user ID if authenticated (optional).

Return type:

UUID | None

async veupath_chatbot.platform.security.get_current_user(user_id)

Get current user ID (required).

Return type:

UUID

veupath_chatbot.platform.security.create_user_token(user_id, expires_in=86400)

Create a signed JWT for the given user.

Parameters:

• user_id (UUID) – User UUID.

• expires_in (int) – Token expiry in seconds (default: 86400).

Return type:

str

Logging

Purpose: Structured logging. get_logger returns a logger configured for JSON/structlog output. Used by all modules.

Key function: get_logger()

Structured logging configuration.

veupath_chatbot.platform.logging.add_request_id(logger, _method_name, event_dict)

Add request ID to log entries if available.

Parameters:

• logger (Logger) – Logger instance.

• _method_name (str) – Method name (structlog processor convention).

• event_dict (MutableMapping[str, Any]) – Log event dictionary to modify.

Returns:

Updated event dict.

Return type:

MutableMapping[str, Any]

veupath_chatbot.platform.logging.setup_logging()

Configure structured logging.

veupath_chatbot.platform.logging.get_logger(name)

Get a structlog logger.

Parameters:

name (str) – Logger name (typically __name__).

Returns:

Configured bound logger.

Return type:

BoundLogger

Context

Purpose: Context variables for request-scoped state. Auth tokens, user IDs, and other per-request data propagated via contextvars.

Context variables for request-scoped data.

Events

Purpose: Application event bus for cross-cutting concerns and inter-service communication.

Event sourcing core: emit events to Redis + project to PostgreSQL.

async veupath_chatbot.platform.events.emit(redis, stream_id, operation_id, event_type, event_data, *, session=None)

Append an event to a Redis Stream and optionally project to PostgreSQL.

Returns the Redis entry ID (e.g. ‘1709234567890-0’).

Return type:

str

async veupath_chatbot.platform.events.read_stream_messages(redis, stream_id)

Read all user + assistant messages from a Redis stream.

Aggregates metadata from surrounding events (tool_call_start/end, citations, planning_artifact, reasoning, subkani events) into each assistant_message so the full conversation context survives refresh.

Used by the GET /strategies/{id} endpoint to return chat history.

Return type:

list[JSONObject]

async veupath_chatbot.platform.events.read_stream_thinking(redis, stream_id)

Derive in-progress thinking state from stream events.

Thinking = tool_call_start events without matching tool_call_end, from the most recent active operation.

Return type:

JSONObject | None

Health

Purpose: Health check logic and readiness probe implementation.

Health-check probes for external dependencies.

async veupath_chatbot.platform.health.check_database(session)

Return True if the database responds to a simple query.

Return type:

bool

async veupath_chatbot.platform.health.check_qdrant()

Return True if Qdrant is reachable and lists collections.

Return type:

bool

Redis

Purpose: Redis client management, connection pooling, and utilities.

Redis connection management for event sourcing.

async veupath_chatbot.platform.redis.init_redis()

Initialize the Redis connection pool.

Return type:

Redis

veupath_chatbot.platform.redis.get_redis()

Get the Redis client. Must call init_redis() first.

Return type:

Redis

async veupath_chatbot.platform.redis.close_redis()

Close the Redis connection pool.

Store

Purpose: Generic store abstractions for in-memory + persistence patterns.

Generic write-through store: in-memory cache + fire-and-forget DB persistence.

Provides the shared save/get/delete/aget/adelete logic so that concrete stores only need to supply their ORM model class, row conversion functions, and custom listing methods.

Subclasses define three class-level attributes:

• _model — SQLAlchemy ORM model class (e.g. ExperimentRow)

• _to_row — callable mapping entity -> dict[str, object] for upsert

• _from_row — callable mapping row -> entity to reconstruct the domain object

The base class derives persist / load / delete from those, eliminating the boilerplate that was previously duplicated across every concrete store.

class veupath_chatbot.platform.store.Identifiable(*args, **kwargs)

Bases: Protocol

Any entity with a string id.

property id: str

__init__(*args, **kwargs)

class veupath_chatbot.platform.store.WriteThruStore

Bases: Generic

In-memory cache backed by fire-and-forget DB writes.

Subclasses must set three class-level attributes:

• _model — SQLAlchemy ORM model (must have an id column)

• _to_row — (entity) -> dict of column values for upsert

• _from_row — (row) -> T to reconstruct the entity from a DB row

Every entity must satisfy the Identifiable protocol (have id: str).

__init__()

save(entity)

get(entity_id)

Return type:

T | None

delete(entity_id)

Return type:

bool

async aget(entity_id)

Return type:

T | None

async adelete(entity_id)

Return type:

bool

Tasks

Purpose: Background task infrastructure and management.

Helpers for fire-and-forget asyncio background tasks.

asyncio.create_task returns a Task that must be stored in a strong reference; otherwise the garbage collector can cancel the task mid-execution. The spawn helper below retains every task in a module-level set until it finishes.

See: https://docs.python.org/3/library/asyncio-task.html#creating-tasks

veupath_chatbot.platform.tasks.spawn(coro, *, name=None)

Schedule coro as a background task with reference retention.

The returned Task is kept alive until completion, after which it is automatically discarded.

If no event loop is running (e.g. called from a sync context outside of async), the coroutine is closed and None is returned.

Return type:

Task[Any] | None

Tool Errors

Purpose: Tool-specific error formatting and handling utilities.

Helpers for standardized AI tool error payloads.

veupath_chatbot.platform.tool_errors.tool_error(code, message, **details)

Build a standardized tool error payload.

Parameters:

• code (str | Enum) – Error code (string or Enum).

• message (str) – Error message.

• details (JSONValue) – Additional details as keyword arguments.

Returns:

Standardized error payload dict.

Return type:

JSONObject

Parsing

Purpose: Input parsing utilities for request processing.

Parsing helpers shared across layers.

veupath_chatbot.platform.parsing.parse_jsonish(value)

Parse tool results that may be JSON or a Python literal.

Some tool frameworks return dict/list, others return a string (often JSON), and some return a Python-literal string representation. This function handles those cases without making assumptions about the payload schema.

Parameters:

value (str | JSONObject | JSONArray | None) – str | JSONObject | JSONArray | None.

Return type:

JSONObject | JSONArray | None

Pydantic Validation

Purpose: Pydantic validation helpers and custom validators.

Utilities for parsing Pydantic validation error text.

Some tool frameworks (including Kani) surface tool-argument validation failures as plain text (Pydantic’s human-readable format) rather than structured JSON. This module provides a best-effort parser so we can return consistent, machine-readable error payloads to the client.

veupath_chatbot.platform.pydantic_validation.parse_pydantic_validation_error_text(text)

Parse Pydantic v2 ValidationError string into a structured payload.

Returns a dict with keys: - model: string (best-effort) - errorCount: int | None - errors: list[dict] (best-effort) - raw: original text

Parameters:

text (str | None) – Pydantic error text (or None).

Returns:

Parsed validation summary or None.

Return type:

JSONObject | None