A general-purpose Python toolkit of powerful, reusable building blocks — the foundation the entire XGen platform is built on.
xgen-sdk is a curated library of strong, general-purpose Python methods and
modules — the kind of logic every serious backend, agent runtime, or RAG
pipeline ends up reinventing. Database pools, config management, object
storage, ABAC authorization, quota evaluation, structured logging, LLM model
discovery, in-app notifications — all consolidated into one canonical
implementation that any Python project can pick up and use.
The XGen platform itself (xgen-core, xgen-workflow, xgen-documents, …)
is essentially a thin layer of business logic built on top of this SDK.
The SDK is where the heavy lifting lives.
One install. No optional extras. Everything you need, ready on
pip install xgen-sdk.
xgen-sdk follows three rules:
- Generalize relentlessly. If a piece of logic is useful in more than one place, it belongs in the SDK — not duplicated in each service.
- Be opinionated where it matters, flexible where it doesn't. Sensible defaults from environment variables, escape hatches for everything.
- Stay batteries-included. No
[extras]to remember, no optional dependency dance.pip install xgen-sdkand you have the full toolkit.
The platform services are intentionally small. xgen-core is admin and auth
business logic. xgen-workflow is workflow orchestration business logic.
xgen-documents is document-processing business logic. All of the heavy
infrastructure code lives here.
| Module | Purpose |
|---|---|
xgen_sdk.db |
psycopg3 connection pool, model-based + table-name CRUD, raw SQL, retries |
xgen_sdk.config |
Redis-backed config with local-file fallback, typed BaseConfig registration |
xgen_sdk.storage |
MinIO / S3-compatible client — upload, download, listing, copy, presigned URLs |
xgen_sdk.auth |
Gateway header parsing, ABAC permissions with wildcards, FastAPI Depends() guards |
xgen_sdk.redis |
General-purpose Redis client for sessions, caching, and pub/sub |
xgen_sdk.logging |
BackendLogger for structured, DB-persisted application logs |
xgen_sdk.quota |
Pure-Python quota policy specs and evaluation (no DB / HTTP coupling) |
xgen_sdk.notification |
Generic per-user persistent in-app notifications with read tracking |
xgen_sdk.llm_catalog |
Dynamic model list for OpenAI / Anthropic / Gemini with TTL cache and fallback |
xgen_sdk.harness |
Built-in agent engine (10-stage pipeline, LangChain-free) — keys from config, sessions in the DB, events to logging, add/remove steps |
xgen_sdk.XgenApp |
One-call bootstrap that wires DB + Config + Storage together |
More general-purpose utilities are added with every release — tracing, retry strategies, schema migration, agent and RAG primitives, prompt templating, async task patterns. If it's reusable, it belongs here.
pip install xgen-sdkRequires Python 3.11+. That's it — every module is installed and ready to import. No extras, no flags, no surprises.
The fastest way to use xgen-sdk inside a FastAPI service:
from contextlib import asynccontextmanager
from fastapi import FastAPI
from xgen_sdk import XgenApp
xgen = XgenApp()
@asynccontextmanager
async def lifespan(app: FastAPI):
xgen.boot() # Connect DB, Config, Storage
app.state.xgen = xgen
yield
xgen.shutdown() # Cleanly release all resources
app = FastAPI(lifespan=lifespan)
@app.get("/users/{user_id}")
async def get_user(user_id: int):
return xgen.db.find_by_id_table("users", user_id)XgenApp.boot() reads its configuration from environment variables
(see Environment below) and initializes:
- PostgreSQL — psycopg3 connection pool via
xgen.db - Config —
RedisConfigManager(or local-file fallback) viaxgen.config - Storage — MinIO client via
xgen.minio_client
Each subsystem can be disabled independently:
xgen = XgenApp(enable_db=True, enable_config=True, enable_storage=False)You're not required to use XgenApp — every module is fully usable on its
own. XgenApp is the convenient default for "I want a FastAPI service with
the standard infrastructure stack."
A psycopg3-based connection pool with both model-class and table-name APIs.
from xgen_sdk.db import XgenDB, database_config
db = XgenDB(database_config)
db.initialize_connection()
# Model-based CRUD
record_id = db.insert(user_model)
records = db.find_by_condition(UserModel, {"status": "active"}, limit=100)
# Table-name CRUD (no model class required)
db.insert_record("users", {"name": "Ada", "email": "ada@example.com"})
rows = db.find_records_by_condition("users", {"is_active": True})
# Raw SQL with parameter binding
rows = db.execute_raw_query(
"SELECT id, name FROM users WHERE created_at > %s", (cutoff,)
)A two-tier configuration system: Redis is the source of truth, with local-file
fallback when Redis is unavailable. Typed config is declared with BaseConfig
and assembled by ConfigComposer.
from xgen_sdk.config import create_config_manager, get_config_value
config = create_config_manager(db_manager=db)
api_key = get_config_value("OPENAI_API_KEY")A MinIO client that covers the full surface area used across XGen services: upload, download, existence checks, recursive listing, copying, and presigned URL generation.
from xgen_sdk.storage import (
upload_file, download_file, get_presigned_url, list_files_in_path,
)
upload_file("documents", "reports/q4.pdf", "/tmp/q4.pdf")
url = get_presigned_url("documents", "reports/q4.pdf", expires=3600)ABAC-style permission system with FastAPI integration. Permissions support
wildcards (admin.*:*, workflow:*) and superusers bypass all checks.
The decorator pattern doubles as a self-registering permission catalog —
the act of declaring an endpoint is the act of declaring its permission.
from fastapi import Depends, Request
from xgen_sdk.auth import require_perm, require_any_perm, require_superuser
@router.get("/roles")
async def list_roles(
request: Request,
session=Depends(require_perm("admin.role:read", description="List roles")),
):
...
@router.delete("/system/reset")
async def reset(_=Depends(require_superuser())):
...At startup, call validate_and_sync(db) to push all decorator-declared
permissions into the database — the frontend then reads them back to build
its role-assignment UI. No more drift between code and DB.
A BackendLogger that writes structured logs to the application database,
tied to the current user and request.
from xgen_sdk.logging import create_logger
log = create_logger(db, user_id=session["user_id"], request=request)
log.info("Workflow started", metadata={"workflow_id": wf_id})
log.success("Workflow finished")
log.error("Step failed", exception=exc)Pure-Python policy evaluation for token quotas. No DB, no HTTP — just dataclasses and functions, so it can be imported safely from any service.
from xgen_sdk.quota import (
resolve_effective_policies, evaluate_quota, period_bounds_kst,
)
effective = resolve_effective_policies(user_id, role_ids, policies)
decision = evaluate_quota(effective, usage_by_period)
if not decision.allowed:
raise HTTPException(429, detail=decision.reason)Generic per-user, persistent, in-app notifications. Failures in the notification path never break the caller's main flow.
from xgen_sdk.notification import (
NotificationPayload, NotificationCategory, NotificationSeverity, publish,
)
publish(db, NotificationPayload(
user_id=123,
category=NotificationCategory.QUOTA,
severity=NotificationSeverity.ERROR,
title="Token quota exceeded",
body="You have exceeded your monthly limit of 1,000,000 tokens.",
link="/mypage?tab=quota",
metadata={"policy_id": 7, "used": 1234567, "limit": 1000000},
))Fetches the live list of available chat / vision models from OpenAI, Anthropic, and Gemini, with an in-process TTL cache and a hardcoded fallback so UI dropdowns never go empty.
from xgen_sdk.llm_catalog import get_models, invalidate
models = get_models(provider="openai", capability="chat")
# [{"id": "gpt-4o", "label": "GPT-4o", ...}, ...]
invalidate("openai") # Call after rotating the API keyThe XGEN agent engine — a 10-stage pipeline (s00…s09) that turns a single
config into a running agent — lives inside the SDK (xgen_sdk.harness), not as
an external dependency. It is dependency-free of LangChain and pure-Python
(httpx only). The engine core stays domain-agnostic; a thin internal integration
layer (xgen_sdk.harness._sdk) wires it into the platform:
- Keys from config — the provider API key is resolved from
xgen_sdk.config(env fallback), so no key needs to be passed by hand. - Sessions in the platform DB —
XgenDBSessionStoreimplements the engine'sSessionStoreprotocol on top ofxgen_sdk.db; multi-turn sessions persist to theharness_sessionstable and resume across processes bysession_id. - Events to logging —
logging_emitterforwards run events to axgen_sdk.logging.BackendLogger.
The pipeline is composable: add_step registers a custom stage through the
engine's extension point; delete_step disables a stage via config. The three
required stages (s01_input, s08_decide, s09_finalize) are protected — the
engine refuses to disable them.
The platform-native entry point is XgenApp.harness(), which returns a Harness
already wired to the app's DB, config, and (optionally) logging:
from xgen_sdk import XgenApp
xgen = XgenApp().boot()
h = xgen.harness(provider="anthropic", model="claude-sonnet-4-6", max_iterations=5)
h.delete_step("s06_context") # remove a non-required step
h.add_step("s_audit", MyAuditStage) # add a custom step (a xgen_sdk.harness.Stage subclass)
print(h.steps()) # active stage ids, in order
state = await h.run("질문", session_id="user-42") # persists + resumes via the DB storeUsed standalone, Harness(...) resolves keys from config the same way and takes
an explicit store= / emitter= for persistence and logging. h.build() returns
the raw engine Pipeline, and the engine types (HarnessConfig,
PipelineBuilder, Stage, SessionStore, ALL_STAGES, REQUIRED_STAGES) are
re-exported from xgen_sdk.harness.
XgenApp.boot() reads these from the environment. All have sensible defaults
for local development.
| Variable | Default | Description |
|---|---|---|
POSTGRES_HOST |
localhost |
DB host |
POSTGRES_PORT |
5432 |
DB port |
POSTGRES_DB |
xgen |
Database name |
POSTGRES_USER |
postgres |
DB user |
POSTGRES_PASSWORD |
postgres |
DB password |
| Variable | Default | Description |
|---|---|---|
REDIS_HOST |
localhost |
Redis host |
REDIS_PORT |
6379 |
Redis port |
REDIS_PASSWORD |
(empty) | Redis password |
| Variable | Default | Description |
|---|---|---|
MINIO_ENDPOINT |
http://minio:9000 |
MinIO endpoint URL |
MINIO_ROOT_USER |
minioadmin |
Access key |
MINIO_ROOT_PASSWORD |
minioadmin |
Secret key |
┌────────────────────────────────────────────────────┐
│ Your application │
│ (xgen-core / xgen-workflow / xgen-documents / │
│ your own agent / RAG / API service) │
│ │
│ ──── business logic only ──── │
└─────────────────────────┬──────────────────────────┘
│
▼
┌────────────────────────────────────────────────────┐
│ xgen-sdk │
│ │
│ db │ config │ storage │ auth │ redis │ logging │
│ quota │ notification │ llm_catalog │ harness │
│ │ XgenApp │ │
│ (+ more in every release) │
└─────────────────────────┬──────────────────────────┘
│
▼
┌────────────────────────────────────────────────────┐
│ PostgreSQL │ Redis │ MinIO │ Gateway │ LLM APIs │
└────────────────────────────────────────────────────┘
Every XGen service speaks directly to the underlying infrastructure through the same SDK code path — no internal HTTP proxies, no duplicated client code. The SDK is the only place those concerns live.
- Python: 3.11, 3.12
- PostgreSQL: 13+
- Redis: 5+
- MinIO: any S3-API-compatible release
psycopg[binary], psycopg-pool, redis, minio, httpx, pydantic,
fastapi, mcp. All installed automatically with pip install xgen-sdk. The
harness engine ships inside the package — no separate engine dependency.
xgen-sdk follows Semantic Versioning. Pin services
with a major-version range to receive bug fixes and additive features
automatically:
dependencies = ["xgen-sdk>=1.14,<2.0"]- Source code: https://github.com/PlateerLab/xgen-sdk
- Issue tracker: https://github.com/PlateerLab/xgen-sdk/issues
- Releases / Changelog: https://github.com/PlateerLab/xgen-sdk/releases
Proprietary — maintained by PlateerLab as the foundation of the XGen platform.