> For the complete documentation index, see [llms.txt](https://docs.citta.ai/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.citta.ai/developers/custom-services/build-a-service-python.md). # Build a service in Python Write a Custom Service in Python with FastAPI and the citta-sdk package - the per-request context, reading and writing data, the session prefix, configuration and local runs. The `citta-sdk` package, imported as `citta`, turns the platform's contract into a few typed objects and a connection pool. Its `citta.service` layer is for Custom Services: it resolves the context of each request, keeps one database pool per application instance, and reads and writes the application's objects the way the platform does. ## Install ```bash pip install 'citta-sdk[fastapi]' # the SDK plus FastAPI and uvicorn pip install 'citta-sdk[aws]' # adds the Lambda adapter pip install citta-sdk # the SDK alone (httpx, psycopg) ``` The package is [`citta-sdk`](https://pypi.org/project/citta-sdk/) on PyPI. It also installs a `citta` command: `citta new hello-world ` writes a complete service with its Lambda image and deploy scripts, the starting point of [Hello world on AWS](/developers/custom-services/hello-world-on-aws.md). `citta new langgraph-agent ` writes a LangGraph agent whose memory lives in the application instance's extension schema, with the same image and deploy scripts; the guide's steps apply to it unchanged. Python 3.10 or newer. A Flask integration, `citta.service.flask`, is installed with `citta-sdk[flask]` and follows the same shape as the FastAPI one shown here. ## The smallest service ```python from fastapi import Depends, FastAPI, Request from fastapi.responses import PlainTextResponse from citta.service import Records, ServiceContext from citta.service.fastapi import citta_context, citta_service, install app = FastAPI() install(app) # reads CITTA_SERVICE_KEY and the other CITTA_* settings @app.get("/{session}", response_class=PlainTextResponse) def hello(request: Request, context: ServiceContext = Depends(citta_context)) -> str: service = citta_service(request) with service.pool(context, read_only=True).connection() as connection: objects = Records(context, connection).objects() return f"hello {context.user.user_name} from {context.tenant.domain}: {', '.join(objects)}" ``` Three calls do the work: * **`install(app)`** attaches a `CittaService` built from the environment and registers the error handlers (below). It returns the service; the pools and clients are closed when the application shuts down. * **`Depends(citta_context)`** resolves the request's `ServiceContext`. It reads the app-instance session from the first path segment and forwards the request's cookie and forwarded headers to the platform's context endpoint, so it works under any route shape - `/{session}` or `/{session}/{rest:path}`. * **`citta_service(request).pool(context)`** is the `psycopg_pool.ConnectionPool` for the request's application instance; `read_only=True` gives the pool on the reader endpoint. Handlers on an event loop use `await service.async_pool(context)`. Every pooled connection has `search_path` set to the three schemas the platform released, extension schema first, so unqualified table names resolve the way the platform intends. ## The context The platform validates the user on every request, so the SDK calls it on every request and never caches the user block. `ServiceContext` is a frozen dataclass: | Attribute | Type | Fields | | ------------------ | ----------------- | ---------------------------------------------------------------------------------------------------------------------------------------------- | | `user` | `User` | `user_id`, `user_name`, `email`, `roles`; `has_role(name)` | | `tenant` | `Tenant` | `tenant_id`, `tenant_name`, `domain` - the tenant whose domain the request came in on | | `app_instance` | `AppInstance` | `application_instance_id`, `agi_id`, `application_data_type_id`, `agi_schema`, `app_schema`, `app_ext_schema` | | `org_context` | `OrgContext` | `org_hierarchy_id`, when the application has an organisation hierarchy | | `connection` | `Connection` | `host` (writer), `read_host` (reader), `port`, `database`, `user`, `password`, `search_path`, `ssl_mode`; the password never appears in `repr` | | `session` | `str` | The app-instance session the request arrived with, verbatim - passed back to the platform in URLs, never parsed | | `sub_path` | `str` | What followed the session in the path | | `session_settings` | `SessionSettings` | `settings`: the PostgreSQL session settings the platform sets on its own connections for this request | The three schemas are the application group instance's data (`agi_schema`), the application's own schema (`app_schema`) and the extension schema the service owns (`app_ext_schema`). `search_path` lists them extension schema first. `app_schema` holds a view of every object the application holds, including objects linked from another application group, and the routines that write them; that is where `Records` reads and writes. `session_settings` matters more than it looks. The platform's object views and merge routines read the acting user, the application instance and the resolved bitemporal properties from session settings. Without them a bitemporal object cannot be read and an insert fails on its audit columns. `Records` applies them with `SET LOCAL` around every read and write; for SQL of your own, use `platform_session`. ## Reading and writing data `Records(context, connection)` reads and writes the application's objects on one released connection, for one request: ```python from citta.service import Records, platform_session with citta_service(request).pool(context).connection() as connection: records = Records(context, connection) records.objects() # ['stay', 'stay_daily_total'] records.count("stay") # rows of the current version, at the request's as-of records.read_relation("stay") # the relation to SELECT from records.write("stay", [{"external_id": "BAY-1001", "property_code": "BAY", "nights": 3, "spend": 840}]) with platform_session(connection, context): rows = connection.execute("SELECT property_code, sum(spend) FROM current_version_stay GROUP BY 1").fetchall() ``` What the released login can do: * **Read** the objects the application holds through their views in `app_schema`, its own objects and those linked from another application group alike. A standard bitemporal or incremental snapshot object is read through its `current_version_` view, which is what `read_relation` and `count` resolve to; a plain object through its view of the same name. Rows marked deleted are not shown. `records.objects()` lists the objects the application holds. * **Write through `Records.write(object, rows, deleted=False)`.** Rows are dicts keyed by field name, with `external_id` for the object's external id. A bitemporal or incremental snapshot object is merged through its `merge_app_bulk_` routine, which writes to the application group that owns the object: an existing external id updates, `deleted=True` marks the rows deleted. A plain object is inserted through its view. An object that can be written neither way raises `NotWritableError`, naming it. Both run inside the platform session settings, so `created_by`, `modified_by` and the bitemporal points are what the platform would record for this user. The call returns one `WrittenRecord` (`id`, `external_id`) per row. * **Own `app_ext_schema`.** Create your own tables there: checkpoints, embeddings, caches. It is first on the search path, so a library that creates tables without a schema name - a LangGraph `PostgresSaver`, a vector store - lands there. The schema travels with the application instance. {% hint style="danger" %} Do not write a bitemporal or incremental snapshot object with raw `INSERT`, `UPDATE` or `DELETE`. Its views are not updatable, and a write that bypassed the merge routine would skip external-id resolution, audit columns and history. `Records.write` is the write path; the object types are described in [Object Types](https://docs.citta.ai/data-modeling-objects/object-types). {% endhint %} Nothing under the platform's own metadata schemas is visible to the login, and no other tenant's data is reachable: the login exists per application instance and is released per authenticated request. ## Storing files A deployed service runs in more than one process - in AWS Lambda, in as many containers as there are concurrent requests - and each has its own disk. A file written while serving one request is not there when the next request reaches another container. Anything that has to outlive a request, such as an uploaded workbook or a result computed from it, goes to the application instance's files. The platform owns that storage. It keeps each file under the tenant and application instance of the request, hands out a short-lived signed URL for every transfer, and stores on the platform's own object storage; the service configures no bucket and holds no storage credentials. ```python from fastapi import Depends, UploadFile from citta.service import AsyncAppInstanceFiles, CittaFileNotFoundError from citta.service.fastapi import citta_files @app.post("/{session}/api/upload") async def upload(file: UploadFile, files: AsyncAppInstanceFiles = Depends(citta_files)): work = files.scoped("services/concierge-agent/") # this service's own folder await work.upload(f"uploads/{file.filename}", await file.read(), file.content_type) return {"stored": file.filename} @app.get("/{session}/api/uploads/{name}") async def read_back(name: str, files: AsyncAppInstanceFiles = Depends(citta_files)): try: return {"bytes": len(await files.scoped("services/concierge-agent/").download(f"uploads/{name}"))} except CittaFileNotFoundError: return {"error": "upload it again"} ``` | Method | Does | | ----------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------- | | `upload(key, content, content_type=None)` | Writes `bytes`, or `str` as UTF-8, replacing whatever was at `key`. | | `download(key)`, `download_text(key)` | The file's bytes, or its text. `CittaFileNotFoundError` when there is no file. | | `list(path="")` | Files and folders directly under `path`, as `FileInfo` (`key`, `file_name`, `size`, `last_modified`, `is_directory`, `content_type`). | | `metadata(key)` | One `FileInfo`. | | `delete(key)` | Deletes the file; deleting nothing is not an error. | | `move(from_key, to_key)` | Copies, then deletes the source, overwriting the destination. Not atomic. | | `scoped(prefix)` | The same interface rooted at a folder: keys in and out are relative to it. | `citta_service(request).files_for(path, headers)` returns the synchronous twin, with the same methods, for code that does not run on the event loop. Keys are relative to the application instance in both directions, so a key from `list` goes straight back into `download` or `move`. A key is a relative path; `..` segments are refused, and so is `.citta/`, the platform's own sidecar store. Four things to know about this storage: * **It is the users' storage too.** The application's file browser shows it, a pipeline's file read node reads from it, and file ingestion writes into it. Keep the service's working files in a folder of its own with `scoped`, and choose one that no pipeline reads. * **Every call acts as the user of the request,** with that user's rights over the application's files, and needs that user's session to be signed in. Work that runs after the request has ended cannot reach the files. * **A user can delete what the service stored.** Treat `CittaFileNotFoundError` as "gone", and ask for the input again rather than failing. * **Each call is a round trip:** one call to the platform, then the transfer itself. Store a few larger objects - one archive of a working folder rather than every file in it - when a request has to restore them before it can answer. A service that keeps a working copy on local disk, for a library that reads files by path, can treat the stored copy as the source of truth: before reading, compare `metadata(key).last_modified` with the copy it unpacked and download again when the stored one is newer; after every write, upload it before the response returns. In local mode (`CITTA_LOCAL_CONTEXT`) the same interface stores in the directory named by `CITTA_LOCAL_FILES_DIR`, so the service code does not change. ## The session prefix The platform strips the reserved segment and forwards `//`, so every path the service receives starts with the user's app-instance session. Routes that know nothing of it - an existing API under `/api`, static files at `/` - would not match. A small ASGI middleware moves the segment out of routing's way while leaving the path the SDK reads intact: {% code title="session\_prefix.py" %} ```python from starlette.types import ASGIApp, Receive, Scope, Send SESSION_PREFIX = "zps" # a platform session string starts with this; nothing else about it is parsed def leading_session(path: str) -> str | None: segments = path.split("/", 2) # "/zpsa_x/api/upload" -> ["", "zpsa_x", "api/upload"] first = segments[1] if len(segments) > 1 else "" return first if first.startswith(SESSION_PREFIX) else None class SessionPrefixMiddleware: def __init__(self, app: ASGIApp) -> None: self.app = app async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: if scope["type"] in ("http", "websocket"): session = leading_session(scope.get("path", "")) if session is not None: prefix = f"/{session}" path = scope["path"] if scope["path"] != prefix else prefix + "/" state = {**(scope.get("state") or {}), "citta_session": session} scope = {**scope, "path": path, "root_path": scope.get("root_path", "") + prefix, "state": state} await self.app(scope, receive, send) ``` {% endcode %} ```python app = FastAPI() app.add_middleware(SessionPrefixMiddleware) install(app) ``` Setting `root_path` to `/` makes Starlette strip it before matching routes and static files, so `/api/health` and `/index.html` match under the platform exactly as they do when the service is run directly; a request with no session in front passes through unchanged. `scope["path"]` keeps the session because `citta_context` reads it from `request.url.path`. The bare launch URL lands on `/` with no trailing slash, which is given a `/` to match. Two consequences for the service's own URLs. Links and redirects must be absolute paths that include the stripped prefix and the session - `X-Forwarded-Prefix` names the prefix - because a relative redirect from `//` would drop the session: ```python def page_url(request: Request, page: str) -> str: prefix = request.headers.get("x-forwarded-prefix", "").rstrip("/") session = getattr(request.state, "citta_session", None) return f"{prefix}/{session}/{page}" if session else f"{prefix}/{page}" ``` And a page served this way should compute its API base from its own location, so that its calls stay under the session. ## Configuration `install(app)` builds the service with `CittaService.from_env()`. Everything is read from the environment, or passed explicitly as a `ServiceSettings`: | Variable | Meaning | | -------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------- | | `CITTA_SERVICE_KEY` | The version's service key, shown once at registration. Required unless in local mode. | | `CITTA_PLATFORM_URL` | The platform origin to call, overriding the request's `X-Forwarded-*` headers. Set it when the service is called directly rather than through the platform. | | `CITTA_LOCAL_CONTEXT` | Path to a JSON context block. When set, no platform call is made (local mode). | | `CITTA_CA_BUNDLE` | A CA file to verify the platform's certificate against, for a platform whose certificate a private CA signed. | | `CITTA_TLS_INSECURE` | `1` disables certificate verification and logs a warning on every client built. Development only. | | `CITTA_POOL_IDLE_SECONDS` | How long an application instance's pool survives without a request. Default 300. | | `CITTA_POOL_MIN_SIZE`, `CITTA_POOL_MAX_SIZE` | Pool bounds. Defaults 1 and 4. | | `CITTA_PLATFORM_TIMEOUT` | Seconds to wait for the platform. Default 30. | | `CITTA_LOCAL_FILES_DIR` | Local mode only: the directory the application instance's files are stored in. Default `.citta-files`. | The service fails at startup, naming the setting, when the key is missing outside local mode or when local mode and a platform URL are both set. ## Errors, and what the browser sees Every failure is a `CittaError` carrying the platform's message verbatim. `install` maps them so that the user and the operator each learn what concerns them: | Raised | Meaning | HTTP answer | | ------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ----------- | | `CittaSessionError` | The cookie is missing or expired, or the user lost access. | 401 | | `CittaServiceConfigError` (with a `kind`) | The platform refused because of the service's or the application's setup: the key authenticates no active version, the key is for another version or service, the instance was deployed before the application pinned the service, or the application pins no service. | 503 | | `RequestShapeError` | The request is not shaped as the platform sends it (no session in the path). | 404 | | Any other `CittaError`, or an unhandled exception | The platform could not be reached or answered something unexpected, or the handler failed. | 522 | The body is `{"error": "", "status": }`, plus `"kind"` on a 503. Register your own exception handler for a class to change its mapping. A 503 is logged with its kind, because it names something an operator has to fix. ## Running locally **Directly, with a copied session.** Run the service on your machine and call it as the platform would, supplying the user's cookie and app-instance session from a browser that is signed in to the platform: open the application, open the developer tools, and from any request to the platform copy the `auth=` value of the `Cookie` header and the session from the request path. Set `CITTA_PLATFORM_URL` to the tenant's URL, since no forwarded headers arrive. ```bash export CITTA_SERVICE_KEY= export CITTA_PLATFORM_URL=https://tenant.example.com uvicorn app:app --port 8000 curl -H 'Cookie: auth=' http://localhost:8000/ ``` The context call reaches the platform over HTTPS; the database connection it returns is reachable only from inside the platform's network, so this route serves data when the service runs in a workspace there. The cookie is the user's session: treat it as a password and let it expire. **Behind the real redirect.** Register a `dev` version whose target serves your build, pin a development instance to it, and open the application's service entry. Everything the platform sends is real, including a different tenant domain per tenant. **Without a platform: local mode.** Put a context block in a file, point its `connection` at a database, and set `CITTA_LOCAL_CONTEXT` to the file. Every request is served as the user in the file; no key, cookie or platform is involved, and the SDK logs one warning at startup so the mode cannot go unnoticed. Files are stored under `CITTA_LOCAL_FILES_DIR`. Local mode replaces only the platform's answer, not the application's schema. `Records` reads a standard bitemporal or incremental snapshot object through its `current_version_` view and writes it through its `merge_app_bulk_` routine, both of which the platform creates when it deploys the application, and both rely on the session settings the platform releases. A database holding plain tables of the same names serves plain objects and your own SQL, but not those objects. The application's schema bundle gives you all of it on a local database: the application's schemas, the owner side of any linked objects, and a context file with the `session` block. See [Hello world on AWS](/developers/custom-services/hello-world-on-aws.md) for downloading and loading it. {% code title="context.local.json" %} ```json { "user": { "userId": 1, "userName": "developer", "email": "developer@example.com", "roles": ["ApplicationUser"] }, "tenant": { "tenantId": 1, "tenantName": "Local", "domain": "local" }, "appInstance": { "applicationInstanceId": 1, "agiId": 1, "applicationDataTypeId": 1, "agiSchema": "agi_local", "appSchema": "app_local", "appExtSchema": "app_ext_local" }, "orgContext": { "orgHierarchyId": null }, "connection": { "host": "localhost", "readHost": "localhost", "port": 5432, "database": "citta_local", "user": "app_svc_local", "password": "change-me", "searchPath": ["app_ext_local", "app_local", "agi_local"], "sslMode": "prefer" } } ``` {% endcode %} ```bash CITTA_LOCAL_CONTEXT=./context.local.json uvicorn app:app --port 8000 curl http://localhost:8000/any-session ``` The same service code runs unchanged in all three, and in AWS Lambda; see [Deploy to AWS](/developers/custom-services/deploy-to-aws.md).