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.comfyui |
ComfyUI API-workflow rules (parameter mapping, validation, tool schema) and an async client |
xgen_sdk.control_policy |
PII and forbidden-word policies: scope switches, masking rules, file byte masking, policy event log |
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)Client uploads reach services through the backend-gateway's resumable chunked
upload, up to 1 GiB per file. Never await file.read() a whole upload: spool it
to disk and work with the path. Downloads stream from a temp file instead of
holding the object in memory.
from xgen_sdk.storage.transfer import spooled, upload_max_bytes, storage_file_response
async with spooled(file, upload_max_bytes()) as (path, size): # 413 past the limit
upload_file(client, path, object_name, bucket_name=bucket)
return await storage_file_response(client, object_name, bucket_name=bucket,
filename="보고서.pdf", media_type="application/pdf")XGEN_UPLOAD_MAX_BYTES (default 1 GiB) is the platform-wide limit and matches the
gateway's UPLOAD_MAX_FILE_BYTES.
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, FastAPI, Request
from xgen_sdk.auth import PrincipalMiddleware, require_perm, require_any_perm, require_superuser
app = FastAPI()
app.add_middleware(PrincipalMiddleware)
@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.
Gateway identity headers are accepted only when their exact values are bound to
an RS256 X-Xgen-Principal assertion. Configure every receiving service with
the Gateway's public key in XGEN_GATEWAY_PUBLIC_KEY_PEM; only the Gateway holds
the corresponding private key. Set XGEN_SERVICE_NAME to the exact audience
that the Gateway uses for that service; broad or missing audiences are rejected.
The single-key variable remains available for compatibility. For rotation, set
XGEN_GATEWAY_KEY_ID and replace it with a strict keyring in
XGEN_GATEWAY_PUBLIC_KEYS:
{
"2026-10": {"key": "-----BEGIN PUBLIC KEY-----..."},
"2026-09": {"key": "-----BEGIN PUBLIC KEY-----...", "not_after": 1792454460}
}The current key named by XGEN_GATEWAY_KEY_ID is the sole entry without
not_after. Every retiring key requires that cutoff, and a token's exp may
not extend beyond it. Once a keyring is configured, assertions without kid,
unknown keys, and the legacy single-key fallback are rejected. Remove a retired
entry to revoke it immediately.
For scoped service-to-service delegation, issuers use XGEN_SERVICE_NAME and
XGEN_SERVICE_SIGNING_KEY. Set XGEN_SERVICE_KEY_ID while rotating keys;
service assertions then carry that exact kid. Receivers configure an explicit
JSON trust map. The legacy single-key form remains supported:
{
"workflow-service": {
"key": "a-distinct-secret-with-at-least-32-bytes",
"permissions": ["document:read", "workflow:execute"]
}
}The rotation form replaces key with keys. Exactly one key is active and has
no cutoff; every other key is retiring and requires not_after:
{
"workflow-service": {
"keys": {
"2026-10": {"key": "current-secret-at-least-32-bytes"},
"2026-09": {"key": "previous-secret-at-least-32-bytes", "not_after": 1792454460}
},
"permissions": ["document:read", "workflow:execute"]
}
}Set that JSON as XGEN_TRUSTED_SERVICES. Service assertions cannot grant
superuser or wildcard permissions, and their declared permissions must be a
subset of the issuer's configured list. Use get_verified_principal_headers()
only to forward an existing verified proof within the same audience. To call a
different backend, use service_principal_headers(..., audience="target-service")
with a user profile obtained from trusted server state.
Document resource grants use a separate, strict HS256 capability codec. These
tokens always use issuer and audience xgen-documents, type
document-capability, and a lifetime of at most 60 seconds. They bind a
deployment, workflow, policy digest, resource id, resource digest, and a unique
non-empty subset of fetch, metadata, access, and download:
from xgen_sdk.auth import sign_document_capability, verify_document_capability
token = sign_document_capability({
"deployment_id": deployment["id"],
"workflow_id": deployment["workflow_id"],
"policy_digest": policy_digest,
"resource_id": resource_id,
"resource_digest": resource_digest,
"operations": ["fetch", "metadata"],
})
claims = verify_document_capability(token)Signing uses XGEN_SERVICE_SIGNING_KEY and XGEN_SERVICE_KEY_ID. Verification
uses that current pair by default. To retain bounded previous keys, set
XGEN_DOCUMENT_CAPABILITY_KEYS to the same {kid: {key, not_after?}} keyring
shape used inside a trusted service entry. A configured capability keyring also
disables fallback to the standalone current key. Use
deployment_policy_digest(row) to bind the fixed deployment authorization
fields; timestamp and unrelated database columns are intentionally excluded,
and a JSON string in public_policy is canonicalized before hashing.
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 keyOne source of truth for turning a ComfyUI API-format workflow into an agent
tool: which inputs are exposed as parameters, how they are validated, the
tool's JSON Schema, and applying arguments. The async client (httpx,
trust_env=False, no redirects) submits, polls history, downloads outputs,
and on cancellation or timeout removes the job from the queue and interrupts it.
from xgen_sdk.comfyui import (
parse_workflow, validate_entry, tool_input_schema, apply_arguments,
ComfyUIClient, run_workflow,
)
graph = parse_workflow(api_export_json) # rejects UI-format JSON
issues = validate_entry(graph, mapping, object_info) # [] means valid
schema = tool_input_schema(mapping)
prompt = apply_arguments(graph, mapping, {"prompt": "a red fox"})
async with ComfyUIClient("http://comfy:8188") as client:
result = await run_workflow(client, prompt, output_node_ids=mapping["outputs"], timeout_s=300)
# {"prompt_id": ..., "duration_ms": ..., "files": [{"node_id", "filename", "content_type", "data"}]}One definition of the admin PII and forbidden-word policies for every service. Each policy kind (PII, forbidden words) has a detection switch and four scope switches: block chat input, mask what goes to the model, strip chat uploads, strip file storage uploads. A scope is active only when both switches are on.
from xgen_sdk import control_policy as cp
rules = cp.load_scope_rules(app_db, cp.SCOPE_CHAT_UPLOAD, workflow_id=wf_id) # per-agent exceptions removed
found = cp.apply_rules(text, rules, mask=False) # detect only
masked = cp.apply_rules(text, rules).text # full or partial masks
result = cp.strip_file(tmp_path, "report.docx", rules) # txt, html, docx/xlsx/pptx, hwp/hwpx, pdf
if result.changed:
size, sha256 = result.size, result.sha256
cp.log_result(app_db, result.found, source=cp.SOURCE_CHAT_UPLOAD, workflow_id=wf_id)PDF needs xgen-pdf at runtime (and chardet helps guess legacy text
encodings). Both are imported lazily, so services that never mask files do not
need them. pdfium is not thread safe, so PDF work runs in small worker
processes that import only this SDK, never the service's own main module.
Hangul documents: HWP 5.0 (compound file) paragraphs, tables, headers, footnotes
and text boxes, plus the preview text and summary information (title, author);
the preview image is replaced with a blank one and the document history is
dropped. Edited paragraphs lose their line layout cache so Hangul lays them out
again. HWPX (zip) is handled the same way. Password, DRM and distribution
documents, HWP 3.0, pictures and scanned PDFs are left as they are
(unmaskable).
| Variable | Default | Meaning |
|---|---|---|
XGEN_FILE_MASKING_MAX_BYTES |
32 MiB | Largest file read into memory; larger files are left as they are (unmaskable) |
XGEN_FILE_MASKING_MAX_UNZIPPED_BYTES |
256 MiB | Largest total unzipped size of an Office file |
XGEN_FILE_MASKING_PDF_WORKERS |
2 | PDF worker processes; 0 runs PDF work in process (tests only) |
XGEN_FILE_MASKING_PDF_TIMEOUT_S |
300 | Time limit for one PDF; the worker is killed and the file is unmaskable |
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 │
│ │ 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, pycryptodome, httpx,
pydantic, fastapi. All installed automatically with pip install xgen-sdk.
xgen-sdk follows Semantic Versioning. Pin services
with a major-version range to receive bug fixes and additive features
automatically:
dependencies = ["xgen-sdk>=2.0,<3.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.