HTTP Server¶

The server framework provides a FastAPI-based implementation of the Haystack HTTP API. Plug in a StorageAdapter and an optional Authenticator and the framework handles routing, content negotiation, authentication, and error wrapping.

Quick Start¶

import uvicorn
from hs_py.fastapi_server import create_fastapi_app
from hs_py.auth_types import StorageAuthenticator
from hs_py.storage.memory import InMemoryAdapter
from hs_py.user import Role, create_user

storage = InMemoryAdapter()

# Create users — StorageAuthenticator reads credentials from the store
import asyncio
asyncio.run(storage.create_user(
    create_user("admin", "s3cret", role=Role.ADMIN)
))

auth = StorageAuthenticator(storage)
app = create_fastapi_app(
    storage=storage,
    authenticator=auth,
    user_store=storage,
)

uvicorn.run(app, host="0.0.0.0", port=8080)

This starts a FastAPI server on port 8080 with SCRAM-SHA-256 authentication, role-based access control, and routes for all standard Haystack ops.

For a simpler setup without user management, use SimpleAuthenticator:

from hs_py.auth_types import SimpleAuthenticator

auth = SimpleAuthenticator({"admin": "secret"})
app = create_fastapi_app(storage=storage, authenticator=auth)

Implementing Operations¶

HaystackOps wraps a StorageAdapter to implement every standard Haystack operation. When you pass a storage argument to create_fastapi_app(), a HaystackOps instance is created automatically.

Haystack Operations¶
Method	HTTP Route	Description
`about()`	`POST /about`	Server information
`ops()`	`POST /ops`	Supported operations (auto-discovered)
`formats()`	`POST /formats`	Supported MIME types
`read(grid)`	`POST /read`	Read records by filter or id
`nav(grid)`	`POST /nav`	Navigate the entity tree
`his_read(grid)`	`POST /hisRead`	Read time-series data
`his_write(grid)`	`POST /hisWrite`	Write time-series data
`point_write(grid)`	`POST /pointWrite`	Write to priority array
`watch_sub(grid)`	`POST /watchSub`	Subscribe to changes
`watch_unsub(grid)`	`POST /watchUnsub`	Unsubscribe points
`watch_poll(grid)`	`POST /watchPoll`	Poll for changes
`invoke_action(grid)`	`POST /invokeAction`	Trigger a named action
`defs(grid)`	`POST /defs`	Query ontology definitions
`libs(grid)`	`POST /libs`	Query ontology libraries

Custom Operations¶

To customise operation behaviour, subclass HaystackOps and pass the instance directly:

from hs_py.ops import HaystackOps
from hs_py import Grid

class MyOps(HaystackOps):
    async def about(self) -> Grid:
        grid = await super().about()
        # Add custom metadata
        return grid

ops = MyOps(storage=storage)
app = create_fastapi_app(ops=ops, authenticator=auth)

Authentication¶

The framework provides two authentication strategies out of the box, and supports custom backends via the Authenticator protocol.

SCRAM-SHA-256 (Password)¶

SimpleAuthenticator accepts a dict[str, str] mapping usernames to passwords. SCRAM keys are pre-derived from ScramCredentials on construction.

from hs_py.auth_types import SimpleAuthenticator

auth = SimpleAuthenticator({"admin": "s3cret", "viewer": "readonly"})

The SCRAM handshake flow is:

Client sends HELLO header with username.
Server responds with handshake token, hash algorithm, and salt.
Client computes SCRAM proof, server verifies.
Server issues a bearer token for subsequent requests.

Security limits:

In-progress handshakes capped at 1,000 to prevent memory exhaustion.
Bearer tokens expire after 3,600 seconds.
Total token count capped at 10,000.

mTLS Client Certificates¶

CertAuthenticator authenticates clients by their TLS client certificate Common Name (CN).

from hs_py.auth_types import CertAuthenticator

# Only accept clients with these CNs
auth = CertAuthenticator(allowed_cns={"device-01", "device-02"})

See TLS and mTLS for setting up mTLS with TLSConfig.

Custom Authenticator¶

Implement the Authenticator protocol for custom backends (LDAP, database, OAuth):

from hs_py.auth_types import Authenticator, ScramCredentials

class MyAuthenticator:
    async def scram_credentials(self, username: str) -> ScramCredentials | None:
        # Look up credentials from your backend
        ...

Storage-Backed Authentication¶

StorageAuthenticator reads SCRAM credentials directly from a UserStore, so user creation and authentication share a single backend:

from hs_py.auth_types import StorageAuthenticator
from hs_py.storage.memory import InMemoryAdapter

storage = InMemoryAdapter()
auth = StorageAuthenticator(storage)
app = create_fastapi_app(
    storage=storage, authenticator=auth, user_store=storage
)

Pass user_store to enable the user management REST API and role enforcement. See Authentication, Users & Permissions for the full details.

Application Setup¶

create_fastapi_app() is the main factory function. It accepts an ops instance (or storage to auto-create one), an optional authenticator, and a URL prefix:

from hs_py.fastapi_server import create_fastapi_app

# Option 1: Pass storage (auto-creates HaystackOps)
app = create_fastapi_app(storage=storage, authenticator=auth)

# Option 2: Pass custom ops instance
app = create_fastapi_app(ops=my_ops, authenticator=auth, prefix="/haystack")

The returned FastAPI application can be extended with additional routes, middleware, and startup hooks:

@app.get("/health")
async def health():
    return {"status": "ok"}

Production Deployment¶

Use uvicorn for production:

uvicorn myapp:app --host 0.0.0.0 --port 8080 --workers 4

Or from Docker:

services:
  server:
    build: .
    ports:
      - "8080:8080"
    environment:
      REDIS_URL: redis://redis:6379
      HS_SUPERUSER_USERNAME: admin
      HS_SUPERUSER_PASSWORD: secret
    depends_on:
      redis:
        condition: service_healthy

Error Handling¶

The ScramAuthMiddleware wraps authentication errors. Operation errors are returned as Haystack error grids automatically.

To return a deliberate error response, raise CallError:

from hs_py.errors import CallError
from hs_py import Grid

class MyOps(HaystackOps):
    async def read(self, grid: Grid) -> Grid:
        filter_str = grid.meta.get("filter")
        if not filter_str:
            raise CallError(
                "missing filter parameter",
                Grid.make_error("missing filter parameter"),
            )
        ...

See Error Handling for error handling patterns.

Watch Push Delivery¶

HaystackOps supports pushing watch updates to subscribed clients via push_watch(). The framework calls set_push_handler() during setup to wire the delivery mechanism (HTTP or WebSocket).

from hs_py.ops import HaystackOps
from hs_py import Grid

class MyOps(HaystackOps):
    async def notify_change(self, watch_id: str, update: Grid) -> None:
        # Push the delta to all subscribed clients
        await self.push_watch(watch_id, update)

WebSocket Endpoint¶

Warning

The WebSocket transport API is experimental and subject to breaking changes in future releases.

The FastAPI app includes a WebSocket endpoint at {prefix}/ws that supports token-based authentication. Clients authenticate via HTTP SCRAM first, then pass the bearer token as the first WebSocket message:

from hs_py import WebSocketClient

async with WebSocketClient("ws://host/api/ws", auth_token=token) as ws:
    about = await ws.about()

For standalone WebSocket servers with SCRAM handshake support, see WebSocketServer and WebSocket Transport.

TLS Configuration¶

For production deployments with TLS:

from hs_py import TLSConfig

tls = TLSConfig(
    certificate_path="server.crt",
    private_key_path="server.key",
    ca_certificates_path="ca.crt",
)

# Use with uvicorn
uvicorn.run(
    app,
    host="0.0.0.0",
    port=8443,
    ssl_certfile="server.crt",
    ssl_keyfile="server.key",
)

See TLS and mTLS for generating test certificates and configuring mTLS.