vs-server

Multi-protocol server framework for Viveka Sutra — HTTP, WebSocket, and SSE on the same class, with pluggable server implementations, lifecycle hooks, and declarative auth guards.

---

Overview

vs-server is the network layer every VS service is built on. It eliminates framework boilerplate — no manual FastAPI() setup, no router wiring, no middleware calls. Controllers are plain classes annotated with @controller. WebSocket and SSE handlers follow the same pattern with @websocket and @sse. A single class can carry all three protocols at once by stacking decorators.

Server implementations are pluggable via VsServerFactory — register any VsServer subclass under a key and retrieve it at startup. VsFastApiServer ships as the built-in implementation and auto-registers as "fastapi" on import.

---

The Problem It Solves

Every service needs routing, middleware, exception handling, and lifecycle management. Without vs-server, each service repeats the same boilerplate.

Without vs-server

from fastapi import FastAPI, APIRouter
from fastapi.middleware.cors import CORSMiddleware
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app):
    await create_tables()
    yield

app = FastAPI(title="my-service", lifespan=lifespan)
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"])

router = APIRouter(prefix="/users")

@router.get("")
async def list_users(): ...

@router.post("")
async def create_user(body: UserRequest): ...

app.include_router(router)
uvicorn.run(app, host="0.0.0.0", port=8000)

With vs-server

from vs_server.decorator.vs_controller_decorator import controller, get, post

@controller("/users")
class UserController:

    @get("")
    async def list_users(self): ...

    @post("")
    async def create_user(self, body: UserRequest): ...

from vs_server.lifecycle.vs_lifecycle import startup
from vs_db.session.vs_db_session_factory import VsDbSessionFactory

config = VsIniConfig("config.ini")
db = VsDbSessionFactory(config)

@startup
async def create_tables():
    await db.create_tables()

server = VsServerFactory.get("fastapi", config)
server.add_controller(UserController())
server.run()

One pattern, every service, zero drift.

---

Installation

pip install vs-server

With FastAPI support:

pip install vs-server[fastapi]

---

Dependencies

Package	Version	Required	Purpose
vs-common	>=0.1.5	Yes	Config, logging, base exceptions
pydantic	>=2.0	Yes	Request/response schema validation
fastapi	>=0.100	No — install with [fastapi] extra	FastAPI HTTP server implementation
uvicorn	>=0.23	No — install with [fastapi] extra	ASGI server
starlette	pulled in by FastAPI	No	CORS middleware
vs-security	any	No — lazy import	JWT auth, filter enrichment

---

Configuration

All server config lives under the [server] section of config.ini.

Key	Default	Description
server.name	vs-server	Server name shown in logs, /, and Swagger
server.version	0.1.0	Version shown in logs, /, and Swagger
server.description	App server developed by Viveka Sutra	Description shown in Swagger
server.base_url	""	Base URL prefix for all core routes (/, /health, /stream-docs)
server.host	127.0.0.1	Host to bind
server.port	8000	Port to bind
server.reload	false	Enable hot reload (dev only)
server.workers	1	Number of worker processes
server.log_level	info	Uvicorn log level
server.docs_enabled	true	Expose /docs, /redoc, /openapi.json, and /stream-docs. Set to false in production to hide all documentation endpoints.
server.ssl_certfile	—	Path to TLS certificate file
server.ssl_keyfile	—	Path to TLS private key file
server.ssl_keyfile_password	—	Password for encrypted private key
api.cors_origins	*	Comma-separated allowed CORS origins

These keys apply to all VsServer implementations. Framework-specific config lives in its own section.

FastAPI-specific config ([fastapi])

Key	Default	Description
fastapi.workers	1	Number of worker processes. Requires fastapi.app_import to be set.
fastapi.reload	false	Enable hot reload (dev only). Requires fastapi.app_import to be set.
fastapi.log_level	info	Uvicorn log level (debug, info, warning, error)
fastapi.app_import	—	Import string for multi-worker or reload mode, e.g. my_service.app:server. Required when workers > 1 or reload = true.

config.ini example for multi-worker production:

[server]
name = my-service
version = 1.0.0
host = 0.0.0.0
port = 8000
docs_enabled = false

[fastapi]
workers = 4
log_level = warning
app_import = my_service.app:server

[api]
cors_origins = https://app.example.com

What is app_import?

When running multiple workers, uvicorn forks child processes that each need to reconstruct the app independently. They can't share the in-memory object from the parent process — they need an import path to build their own copy.

app_import is module.path:variable_name where variable_name is whatever you assigned VsServerFactory.get() to:

# my_service/app.py
server = VsServerFactory.get("fastapi", config)  # → app_import = my_service.app:server
server.run()

If app_import is not set, vs-server will attempt to auto-detect it and log it at startup. Copy the logged value into your config to make it explicit.

Custom server implementations follow the same pattern — add a [my_framework] section in config.ini for any framework-specific keys and read them in your VsServer subclass.

config.ini example:

[server]
name = my-service
version = 1.0.0
host = 0.0.0.0
port = 8000
workers = 2
ssl_certfile = certs/server.crt
ssl_keyfile = certs/server.key

[api]
cors_origins = https://app.example.com,https://admin.example.com

[logging]
level = INFO
file_path = ./logs/service.log

---

Quick Start

from vs_common.config.vs_ini_config import VsIniConfig
from vs_common.log.vs_log_manager import VsLogManager
from vs_common.schema.vs_log_config import VsLogConfig
from vs_server.server.vs_fast_api_server import VsFastApiServer  # noqa — auto-registers "fastapi"
from vs_server.factory.vs_server_factory import VsServerFactory
from vs_server.decorator.vs_server_registry import server_registry

import my_service.controllers  # noqa — triggers @controller registration


@server_registry(hooks="my_service.lifecycle")
def main():
    config = VsIniConfig("config.ini")
    VsLogManager.init(VsLogConfig(level=config.get("logging.level", default="INFO")))

    server = VsServerFactory.get("fastapi", config)
    server.add_controller(UserController())
    server.run()


if __name__ == "__main__":
    main()

---

How It All Fits Together

Application Startup
    └── import VsFastApiServer          # auto-registers "fastapi" into VsServerFactory

@server_registry(hooks="my_service.lifecycle")
    └── imports my_service.lifecycle    # @startup / @shutdown self-register into hook lists

VsServerFactory.get("fastapi", config)
    ├── reads name/version/host/port/TLS from config.ini
    ├── wires CORS, exception handlers, lifespan
    └── exposes /, /health  (+ /stream-docs once a WS/SSE handler is registered)

server.get_app().add_middleware(VsDbMiddleware)   # register manually if using vs-db

server.add_controller(UserController())
    └── reads @controller + @get/@post/... → wires FastAPI router

server.run()
    ├── calls @startup hooks in order
    ├── starts uvicorn
    └── calls @shutdown hooks on stop

---

Standalone Route Functions

HTTP verb decorators can be applied to plain functions outside any class. Decorate the function and place it in a package passed to server_registry(routes=[...]) — no manual registration needed.

# my_service/routes.py
from vs_server.decorator.vs_controller_decorator import get, post, delete

@get("/users", response_model=list[UserResponse])
async def list_users():
    ...

@post("/users", status_code=201, response_model=UserResponse)
async def create_user(body: UserRequest):
    ...

@delete("/users/{user_id}", status_code=204)
async def delete_user(user_id: str):
    ...

@server_registry(
    hooks="my_service.lifecycle",
    routes=["my_service.routes", "my_service.admin.routes"],
)
def main():
    server = VsServerFactory.get("fastapi", config)
    server.run()

The scanner walks each package in routes, finds all functions decorated with @get/@post/@put/@delete/@patch, and wires them into the server automatically at startup.

Guards work the same way as on controller methods:

@get("/admin/report", guards=[require_admin_role])
async def get_report():
    ...

---

Request and Response Models

All request bodies and response models must be Pydantic models — not SQLAlchemy entities or plain dicts. This is a hard requirement of FastAPI and applies to all server implementations built on vs-server.

from pydantic import BaseModel

class UserRequest(BaseModel):
    name: str
    email: str

class UserResponse(BaseModel):
    id: str
    name: str
    email: str

    model_config = {"from_attributes": True}

The mapping between your SQLAlchemy entity and your Pydantic schema is your responsibility:

Request (Pydantic) → Service → Entity (SQLAlchemy) → DB
DB → Entity (SQLAlchemy) → Service → Response (Pydantic)

Never pass a SQLAlchemy entity as response_model — FastAPI will reject it at startup. Never return a SQLAlchemy entity directly from a route — serialize it first.

Keeping schemas separate from entities is intentional: the DB schema and API schema can evolve independently, and sensitive fields (e.g. password) are excluded from responses by default.

---

HTTP Controllers

Annotate a class with @controller and methods with HTTP verb decorators.

from vs_server.decorator.vs_controller_decorator import controller, get, post, put, delete, patch
from pydantic import BaseModel

class UserRequest(BaseModel):
    name: str
    email: str

@controller("/users", description="User management endpoints")
class UserController:

    @get("", response_model=list[UserResponse])
    async def list_users(self): ...

    @get("/{user_id}", response_model=UserResponse)
    async def get_user(self, user_id: str): ...

    @post("", status_code=201, response_model=UserResponse)
    async def create_user(self, body: UserRequest): ...

    @put("/{user_id}", response_model=UserResponse)
    async def update_user(self, user_id: str, body: UserRequest): ...

    @delete("/{user_id}", status_code=204)
    async def delete_user(self, user_id: str): ...

Register via server.add_controller():

server.add_controller(UserController())

All FastAPI route parameters (response_model, status_code, tags, summary, etc.) pass through via **kwargs.

@controller parameters

Parameter	Type	Required	Default	Description
path	str	Yes	—	URL prefix for all routes in this controller
description	str	No	""	Tag description shown in Swagger
guards	List[Callable]	No	[]	Applied to every route in this controller

HTTP verb decorators

Decorator	HTTP Method
@get(path, summary, description, guards=[], **kwargs)	GET
@post(path, summary, description, guards=[], **kwargs)	POST
@put(path, summary, description, guards=[], **kwargs)	PUT
@delete(path, summary, description, guards=[], **kwargs)	DELETE
@patch(path, summary, description, guards=[], **kwargs)	PATCH

Guards on a method are merged with controller-level guards — both run.

---

WebSocket

Annotate a class with @websocket and message handlers with @on_message.

from vs_server.decorator.vs_websocket_decorator import websocket, on_message

@websocket("/ws/chat", description="Real-time chat over WebSocket")
class ChatHandler:

    @on_message(
        description="Handle incoming chat message",
        args={"text": "Message text", "room": "Target room ID"},
    )
    async def handle_message(self, data: dict):
        text = data.get("text")
        # process and send response

    @on_message(description="Handle ping")
    async def handle_ping(self, data: dict):
        ...

Register via server.add_websocket():

server.add_websocket(ChatHandler())

Connection lifecycle: Guards run at connect time. If any guard raises, the connection is closed with code 1008 before accept(). Message guards run per-message before the handler.

@websocket parameters

Parameter	Type	Required	Default	Description
path	str	Yes	—	WebSocket URL path
description	str	No	""	Shown in /stream-docs
guards	List[Callable]	No	[]	Run at connection time — signature: async guard(websocket)

@on_message parameters

Parameter	Type	Required	Default	Description
description	str	No	""	Shown in /stream-docs
args	dict	No	{}	Argument descriptions shown in /stream-docs
guards	List[Callable]	No	[]	Run per-message — signature: async guard(websocket, data)

---

Server-Sent Events (SSE)

Annotate a class with @sse and the stream method with @sse_stream.

from fastapi import Request
from vs_server.decorator.vs_sse_decorator import sse, sse_stream

@sse("/events/notifications", description="Notification stream")
class NotificationStream:

    @sse_stream(
        description="Stream live notifications",
        args={"user_id": "Filter notifications for this user"},
    )
    async def stream(self, request: Request):
        while not await request.is_disconnected():
            event = await notification_queue.get()
            yield event

Register via server.add_sse():

server.add_sse(NotificationStream())

The stream method must be an async generator. Each value yielded is sent as a data: <value>\n\n SSE event.

VsSseEvent — named events and resumable streams

A bare yield sends a lone data: line, which is all most streams need. Yield a VsSseEvent instead when the client has to dispatch on the kind of frame, or reconnect without losing its place.

from vs_server.decorator.vs_sse_decorator import sse, sse_stream
from vs_server.schema.vs_sse_event import VsSseEvent

@sse("/v1/agent/stream", description="Agent analysis and answers")
class AgentStream:

    @sse_stream(description="Everything this user's agents are saying")
    async def stream(self, request: Request):
        async for frame in read_user_stream(request):
            yield VsSseEvent(
                data=frame.payload,     # dict or list — JSON-encoded for you
                event=frame.kind,       # "analysis" or "response"
                id=frame.stream_id,     # what Last-Event-ID sends back on reconnect
            )

Field	Type	Sent as	Purpose
data	Any	data:	The payload. Strings go as-is, anything else is JSON-encoded. Multi-line payloads become one data: line each, per the spec.
event	str	event:	Names the frame, so addEventListener("analysis", ...) works
id	str	id:	The browser echoes the last one back as Last-Event-ID when it reconnects
retry_ms	int	retry:	How long the client waits before reconnecting
comment	str	:	Ignored by clients — used for keep-alives

On the client, the frames arrive already separated:

const stream = new EventSource("/v1/agent/stream");
stream.addEventListener("analysis", e => showThinking(JSON.parse(e.data)));
stream.addEventListener("response", e => showAnswer(JSON.parse(e.data)));

Keep-alives. VsSseEvent.keep_alive() yields a comment-only frame. Proxies close connections that look idle, and a stream waiting on a slow agent looks exactly like one — send one every 15–30 seconds:

try:
    frame = await asyncio.wait_for(queue.get(), timeout=20)
    yield VsSseEvent(data=frame.payload, event=frame.kind, id=frame.id)
except asyncio.TimeoutError:
    yield VsSseEvent.keep_alive()

Resuming. Give each frame an id and a reconnecting browser sends Last-Event-ID automatically. Read it off the request and replay from that point rather than from the beginning:

async def stream(self, request: Request):
    resume_from = request.headers.get("Last-Event-ID")
    async for frame in read_user_stream(request, after=resume_from):
        yield VsSseEvent(data=frame.payload, event=frame.kind, id=frame.id)

Mixing the two styles is fine — a stream can yield plain strings and VsSseEvents in the same generator.

@sse parameters

Parameter	Type	Required	Default	Description
path	str	Yes	—	SSE endpoint URL path
description	str	No	""	Shown in /stream-docs
guards	List[Callable]	No	[]	Run before the stream starts

@sse_stream parameters

Parameter	Type	Required	Default	Description
description	str	No	""	Shown in /stream-docs
args	dict	No	{}	Argument descriptions shown in /stream-docs
guards	List[Callable]	No	[]	Run before the stream starts

---

Multi-Protocol Classes

A single class can carry HTTP, WebSocket, and SSE by stacking decorators:

from vs_server.decorator.vs_controller_decorator import controller, get
from vs_server.decorator.vs_websocket_decorator import websocket, on_message
from vs_server.decorator.vs_sse_decorator import sse, sse_stream

@controller("/chat", description="Chat HTTP endpoints")
@websocket("/ws/chat", description="Chat WebSocket")
@sse("/events/chat", description="Chat event stream")
class ChatHandler:

    @get("/history")
    async def get_history(self): ...

    @on_message(description="Send a message")
    async def handle_message(self, data: dict): ...

    @sse_stream(description="Live chat events")
    async def stream(self, request): ...

Register each protocol separately:

handler = ChatHandler()
server.add_controller(handler)
server.add_websocket(handler)
server.add_sse(handler)

---

Guards

Guards are async callables that run before a route, WebSocket connection, or SSE stream. If a guard raises, the request is rejected.

from vs_security.guard.vs_security_factory import VsSecurityFactory

# controller-level guard — applies to all routes
@controller("/admin", guards=[VsSecurityFactory.get()])
class AdminController:

    # route-level guard — merged with controller guards
    @get("/report", guards=[require_admin_role])
    async def get_report(self): ...

Guard signature by protocol:

Protocol	Point	Guard signature
HTTP	Per-request	async def guard() -> None (FastAPI Depends)
WebSocket	At connect	async def guard(websocket: WebSocket) -> None
WebSocket	Per-message	async def guard(websocket: WebSocket, data: dict) -> None
SSE	Before stream	async def guard() -> None (FastAPI Depends)

Guards that raise PermissionError are handled by the built-in exception handler and return a 403 response on HTTP. On WebSocket, the connection is closed with code 1008.

---

Lifecycle Hooks

Use @startup and @shutdown to run code when the server starts and stops.

from vs_common.config.vs_ini_config import VsIniConfig
from vs_server.lifecycle.vs_lifecycle import startup, shutdown
from vs_db.session.vs_db_session_factory import VsDbSessionFactory

config = VsIniConfig("config.ini")
db = VsDbSessionFactory(config)

@startup
async def create_tables():
    await db.create_tables()

@shutdown
async def close_connections():
    await db.close()

Hooks self-register into global lists at import time. They run in registration order. Both sync and async hooks are supported.

Register hook modules with server_registry so they are imported before the server starts:

from vs_server.decorator.vs_server_registry import server_registry

@server_registry(hooks="my_service.lifecycle")
def main():
    ...

VsServer.startup() iterates _startup_hooks. VsServer.shutdown() iterates _shutdown_hooks.

---

VsServerFactory and @server

VsServerFactory maps string keys to VsServer subclasses. Register once, retrieve by key anywhere.

from vs_server.factory.vs_server_factory import VsServerFactory
from vs_server.server.vs_fast_api_server import VsFastApiServer  # noqa — auto-registers "fastapi"

server = VsServerFactory.get("fastapi", config)

Adding a custom server implementation

from vs_server.server.vs_server import VsServer
from vs_server.decorator.vs_server_decorator import server
from vs_server.factory.vs_server_factory import VsServerFactory


@server("flask")
class VsFlaskServer(VsServer):

    def add_controller(self, instance, guards=None): ...
    def add_router(self, router, prefix=""): ...
    def add_websocket(self, instance): ...
    def add_sse(self, instance): ...
    def get_app(self): ...
    def run(self): ...


VsServerFactory.register("flask", VsFlaskServer)

Or scan the package automatically with server_registry:

@server_registry(server="my_pkg.server")
def main():
    server = VsServerFactory.get("flask", config)
    server.run()

---

server_registry

Scans packages for @server-decorated classes, @startup/@shutdown hook functions, and standalone route functions. Works as a decorator on main() or as a plain call.

from vs_server.decorator.vs_server_registry import server_registry

@server_registry(
    hooks="my_service.lifecycle",
    routes=["my_service.routes", "my_service.admin.routes"],
)
def main():
    server = VsServerFactory.get("fastapi", config)
    server.add_controller(UserController())  # controllers still registered manually
    server.run()

Parameter	Type	Description
server	str	Package path to scan for @server-decorated classes. Not needed for "fastapi" — it auto-registers on import.
hooks	str	Package path to import for @startup/@shutdown registration.
routes	List[str]	Package paths to scan for standalone @get/@post/... decorated functions. Auto-wired into the server at startup — no add_route call needed.

---

HTTPS / TLS

Set ssl_certfile and ssl_keyfile in config.ini. No code changes required.

[server]
ssl_certfile = certs/server.crt
ssl_keyfile = certs/server.key
ssl_keyfile_password = optional-passphrase

TLS is enabled when both ssl_certfile and ssl_keyfile are present. The startup banner prints TLS: enabled.

---

Built-in Endpoints

Every VsFastApiServer instance exposes three endpoints automatically:

Endpoint	Description
GET /	Returns {"service": name, "version": version, "status": "running"}
GET /health	Returns VsServerHealth with overall status, uptime, cache, and dependency health
GET /docs	Swagger UI — hidden when server.docs_enabled = false
GET /redoc	ReDoc UI — hidden when server.docs_enabled = false
GET /stream-docs	Custom HTML docs page for WebSocket and SSE endpoints — registered only once at least one @websocket or @sse handler is added, and hidden when server.docs_enabled = false

If server.base_url is set in config.ini, /, /health, and /stream-docs are prefixed with it.

Set server.docs_enabled = false in production to disable all four documentation endpoints at once.

/health response

{
  "status": "healthy",
  "name": "my-service",
  "version": "1.0.0",
  "uptime_seconds": 142.3,
  "cache": "healthy",
  "instance_id": "hostname",
  "dependencies": {
    "database": "healthy"
  },
  "timestamp": "2026-01-01T00:00:00Z"
}

Overall status logic:

Condition	Status
All dependencies healthy	healthy
Some healthy, some unhealthy	degraded
All unhealthy	unhealthy

Override _check_dependencies() in a VsServer subclass to include custom dependency checks:

async def _check_dependencies(self) -> dict:
    try:
        await db.execute("SELECT 1")
        return {"database": "healthy"}
    except Exception:
        return {"database": "unhealthy"}

---

Documentation

URL	Content
/docs	Swagger UI — HTTP endpoints only (FastAPI native)
/stream-docs	Custom HTML page — WebSocket and SSE endpoints

HTTP endpoints are documented automatically via FastAPI's Swagger UI. WebSocket and SSE endpoints are not supported by OpenAPI — they appear in /stream-docs instead, generated by VsDocsRenderer.

get_docs_url() returns the framework-native docs URL ("/docs" for FastAPI). Override to change or suppress it.

---

Security

vs-server provides the HTTP-side of security — token extraction, middleware wiring, and request abstraction. The auth logic (session validation, JWT, user loading) lives in vs-security.

enable_security()

Call enable_security() after creating the server. It registers a middleware that runs before every request, and automatically configures Swagger UI with a Bearer token "Authorize" button. Only secured endpoints show the lock icon — excluded paths do not.

server = VsServerFactory.get("fastapi", config)
server.enable_security()   # default: VsBearerTokenFilter, all routes secured
server.run()

With excluded paths (always public, no auth required):

server.enable_security(VsBearerTokenFilter(exclude_paths=["/health", "/auth/login", "/auth/refresh"]))

With a custom filter:

server.enable_security(MyFilter(exclude_paths=["/health"]))

To disable Swagger UI entirely (e.g. in production), set server.docs_enabled = false in config.ini.

VsSecurityFilter

Extend this to implement your own auth mechanism — API key, session cookie, mTLS, anything. Call set_auth_context(context) when done.

from vs_server.filter.vs_security_filter import VsSecurityFilter
from vs_server.filter.vs_request import VsRequest
from vs_security.guard.vs_security import set_auth_context

class ApiKeyFilter(VsSecurityFilter):

    async def do_filter(self, request: VsRequest) -> None:
        key = request.get_header("X-Api-Key")
        if not key:
            raise VsAuthenticationError("Missing API key")
        context = await key_repo.validate(key)
        set_auth_context(context)

server.enable_security(ApiKeyFilter(exclude_paths=["/health"]))

process() in the base class handles exclude_paths — your do_filter() only sees requests that need auth.

VsBearerTokenFilter

Built-in filter that extracts Authorization: Bearer <token>, calls VsAuthManager.get_instance().session.validate(token), and sets the context. Used by default when enable_security() is called with no argument.

Requires vs-security and VsAuthManager.init() to be called at startup.

Override enrich() to fetch fresh user details from DB after token validation and add them to the context — without touching every handler:

from vs_server.filter.vs_bearer_token_filter import VsBearerTokenFilter

class MyFilter(VsBearerTokenFilter):

    def __init__(self, user_repo):
        super().__init__(exclude_paths=["/health", "/auth/login"])
        self._user_repo = user_repo

    async def enrich(self, context: AppAuthContext) -> AppAuthContext:
        user = await self._user_repo.find_by_id(context.user_id)
        context.metadata["email"] = user.email
        context.metadata["plan"] = user.plan
        return context

server.enable_security(MyFilter(user_repo=UserRepo()))

enrich() default is a no-op — only override when you need live data beyond what's in the JWT.

VsRequest / VsFastApiRequest

VsRequest is the framework-agnostic request abstraction passed to do_filter(). VsFastApiRequest wraps FastAPI's Request.

Method	Returns	Description
get_header(name)	Optional[str]	Returns the header value, case-insensitive
headers	Dict[str, str]	All request headers
get_path()	str	Request URL path (e.g. /users/me)

Implement VsRequest when adding a non-FastAPI server — wrap the native request object and pass it to VsSecurityFilter.process().

---

Using with vs-db

Initialize VsDbSessionFactory before creating the server, then register VsDbMiddleware explicitly:

from vs_db.session.vs_db_session_factory import VsDbSessionFactory
from vs_db.middleware.vs_db_middleware import VsDbMiddleware
from vs_server.server.vs_fast_api_server import VsFastApiServer  # noqa — auto-registers "fastapi"
from vs_server.factory.vs_server_factory import VsServerFactory

def main():
    config = VsIniConfig("config.ini")
    VsDbSessionFactory(config, entities=["my_service.models"])

    server = VsServerFactory.get("fastapi", config)
    server.get_app().add_middleware(VsDbMiddleware)
    server.add_controller(UserController())
    server.run()

VsDbMiddleware must be registered before server.run() — middleware cannot be added after the server starts accepting requests.

---

Error Handling

vs-server has two tiers of exception handling, both producing the same JSON shape:

• VsAPIException — server-owned exceptions with an explicit error_code and details dict. Use these when you want full control over the error code and extra context.
• VsBaseException subclasses — any exception from any VS library (e.g. VsAuthenticationError from vs-security) is handled automatically via duck-typing. The handler reads status_code if present (defaults to 500) and derives the error code from the class name.

vs-server has no direct dependency on vs-security — the duck-typing approach means all VS library exceptions work automatically.

Raise exceptions from vs_server.schema.exceptions inside controllers — the built-in handlers convert them to structured JSON responses automatically.

from vs_server.schema.exceptions import NotFoundException, ValidationException

@controller("/users")
class UserController:

    @get("/{user_id}")
    async def get_user(self, user_id: str):
        user = await repo.find(user_id)
        if user is None:
            raise NotFoundException(f"User '{user_id}' not found")
        return user

Error response format:

{
  "success": false,
  "error": {
    "code": "NOT_FOUND",
    "message": "User 'abc' not found",
    "details": {}
  }
}

Built-in exception types:

Exception	HTTP Status	Error Code	Default Message
NotFoundException	404	NOT_FOUND	Resource not found
ValidationException	400	VALIDATION_ERROR	Invalid request
UnauthorizedException	401	UNAUTHORIZED	Authentication required
ForbiddenException	403	FORBIDDEN	Access denied
InternalServerException	500	INTERNAL_ERROR	Internal server error
ServiceUnavailableException	503	SERVICE_UNAVAILABLE	Service unavailable
RateLimitException	429	RATE_LIMIT_EXCEEDED	Rate limit exceeded
TimeoutException	504	TIMEOUT_ERROR	Request timed out

All extend VsAPIException, which extends VsBaseException. Catch by either base type to handle errors uniformly.

VsBaseException subclasses are handled automatically:

Any exception that extends VsBaseException (from vs-common) is caught and converted to a JSON error response. The handler uses getattr(exc, 'status_code', 500) — if the exception carries a status_code (e.g. auth errors from vs-security), that is used; otherwise it defaults to 500. The error code is derived from the class name — e.g. VsInvalidCredentialsError → INVALID_CREDENTIALS, VsTokenExpiredError → TOKEN_EXPIRED.

vs-server does not depend on vs-security — the duck-typing approach means any library's exceptions work automatically as long as they extend VsBaseException and optionally set status_code.

@post("/auth/login", response_model=LoginResponse)
async def login(body: LoginRequest):
    # VsInvalidCredentialsError → 401 INVALID_CREDENTIALS automatically
    session = await VsAuthManager.get_instance().authenticate(body.type, body)
    return LoginResponse(access_token=session.access_token, refresh_token=session.refresh_token)

Automatically handled Python exceptions:

Exception	HTTP Status
VsBaseException (and subclasses)	Derived from exc.status_code
PermissionError	403
ValueError	400
fastapi.RequestValidationError	400 (with field-level detail)
Any other Exception	500

---

Extending vs-server

Implement VsServer to add support for any framework. The abstract interface is:

from vs_server.server.vs_server import VsServer
from vs_server.config.vs_base_config import VsBaseConfig

class MyServer(VsServer):

    def __init__(self, config: VsBaseConfig):
        super().__init__(config)

    def add_controller(self, instance, guards=None) -> None: ...
    def add_router(self, router, prefix: str = "") -> None: ...
    def add_websocket(self, instance) -> None: ...
    def add_sse(self, instance) -> None: ...
    def get_app(self): ...
    def run(self) -> None: ...

Override template methods to customise behaviour:

Method	Default	Override to
get_host()	Reads server.host	Hard-code or compute the bind address
get_port()	Reads server.port	Hard-code or compute the bind port
get_reload()	FastAPI only — reads fastapi.reload	Force hot-reload on/off
get_workers()	FastAPI only — reads fastapi.workers	Set worker count programmatically
get_log_level()	FastAPI only — reads fastapi.log_level	Set log level programmatically
get_app_import()	FastAPI only — reads fastapi.app_import	Return import string for multi-worker mode
get_ssl_certfile()	Reads server.ssl_certfile	Load cert from a secret store
get_ssl_keyfile()	Reads server.ssl_keyfile	Load key from a secret store
get_ssl_keyfile_password()	Reads server.ssl_keyfile_password	Load password from a secret store
is_docs_enabled()	Reads server.docs_enabled	Toggle docs based on environment or role
startup()	Logs + runs @startup hooks	Add custom startup logic. Call super().startup() to keep hooks.
shutdown()	Runs @shutdown hooks + logs	Add custom shutdown logic. Call super().shutdown() to keep hooks.
get_docs_url()	Returns None	Return the framework-native docs URL
_check_dependencies()	Returns {}	Return dict[str, "healthy" | "unhealthy"] merged into /health
_register_middleware(app)	Adds CORS	Add custom middleware. Call super() to keep CORS.
_register_exception_handlers(app)	Adds built-in handlers	Add custom exception handlers. Call super() to keep built-ins.

---

Class Reference

---

VsServer

Abstract base class for all server implementations. Manages config, logging, and lifecycle hooks.

Constructor:

Parameter	Type	Description
config	VsBaseConfig	Application config — reads server.* keys

Attributes set from config:

Attribute	Config key	Default
name	server.name	vs-server
version	server.version	0.1.0
description	server.description	App server developed by Viveka Sutra
base_url	server.base_url	""

Config methods — read from config.ini by default. Override any of them in a subclass to change the value without touching config:

Method	Default config key	Default value	Description
get_host()	server.host	127.0.0.1	Bind address
get_port()	server.port	8000	Bind port
get_reload()	server.reload	false	Hot-reload (dev only)
get_workers()	server.workers	1	Worker process count
get_log_level()	server.log_level	info	Uvicorn log level
get_ssl_certfile()	server.ssl_certfile	None	TLS certificate path
get_ssl_keyfile()	server.ssl_keyfile	None	TLS private key path
get_ssl_keyfile_password()	server.ssl_keyfile_password	None	TLS key password
is_docs_enabled()	server.docs_enabled	true	Whether to expose documentation endpoints

Lifecycle methods:

Method	Signature	Description
startup	async startup() -> None	Logs startup banner, then runs all @startup hooks. Call super().startup() to keep.
shutdown	async shutdown() -> None	Runs all @shutdown hooks, then logs stop. Call super().shutdown() to keep.
get_docs_url	get_docs_url() -> Optional[str]	Returns None. Override to return framework-native docs URL.
add_controller	add_controller(instance, guards=None) -> None	Abstract. Register a @controller instance.
add_router	add_router(router, prefix="") -> None	Abstract. Register a raw framework router.
add_websocket	add_websocket(instance) -> None	Abstract. Register a @websocket instance.
add_sse	add_sse(instance) -> None	Abstract. Register an @sse instance.
get_app	get_app() -> Any	Abstract. Return the underlying framework app.
run	run() -> None	Abstract. Start the server.

---

VsServerFactory

Registry mapping string keys to VsServer subclasses. Thread-safe class-level dict.

Methods:

Method	Signature	Description
register	register(key: str, server_class: Type[VsServer]) -> None	Register a server class. Raises TypeError if not a VsServer subclass.
get	get(key: str, config: VsBaseConfig) -> VsServer	Instantiate and return a server. Raises KeyError if key not registered.

Notes:

• VsFastApiServer auto-registers as "fastapi" when its module is imported.
• Import VsFastApiServer before calling VsServerFactory.get("fastapi", config).

---

VsFastApiServer

Built-in VsServer implementation backed by FastAPI and uvicorn. Auto-registers as "fastapi" on import.

Behaviour at construction:

• Raises ImportError if fastapi is not installed.
• Builds the FastAPI app with lifespan, CORS middleware, exception handlers, and core routes.

Built-in routes: GET /, GET /health (and GET /stream-docs once a @websocket/@sse handler is added).

Methods (beyond VsServer contract):

Method	Description
enable_security(filter?, exclude_paths?)	Register security middleware. Defaults to VsBearerTokenFilter. Pass a custom VsSecurityFilter subclass for any other mechanism.
get_docs_url()	Returns "/docs"
_check_dependencies()	Override to add custom dependency health checks
_register_middleware(app)	Adds CORS. Call super() to keep it.
_register_exception_handlers(app)	Adds built-in handlers. Call super() to keep them.

Notes:

• reload=True and workers > 1 are mutually exclusive. If reload is true, workers is ignored.
• TLS is enabled when both ssl_certfile and ssl_keyfile are set.
• VsDbMiddleware must be registered manually: server.get_app().add_middleware(VsDbMiddleware).

---

@server

Decorator. Marks a class as a named server implementation. Validates that it extends VsServer.

Parameters:

Parameter	Type	Required	Description
key	str	Yes	Registry key used with VsServerFactory.get(key, config)

---

server_registry

Scans packages for @server-decorated classes and @startup/@shutdown hook functions. Works as a function decorator or a plain call. All parameters are optional.

Parameters:

Parameter	Type	Description
server	Optional[str]	Package path to scan for @server classes
hooks	Optional[str]	Package path to import for lifecycle hook registration

---

@startup / @shutdown

Decorators. Register a function into _startup_hooks or _shutdown_hooks at import time. Both sync and async functions are supported.

from vs_server.lifecycle.vs_lifecycle import startup, shutdown

@startup
async def on_start(): ...

@shutdown
def on_stop(): ...

Hooks run in registration order via VsServer.startup() and VsServer.shutdown().

---

@controller

Decorator. Marks a class as an HTTP controller. Sets cls._vs_controller with path, description, and guards.

Parameters:

Parameter	Type	Required	Default	Description
path	str	Yes	—	URL prefix for all routes in this controller
description	str	No	""	Tag description shown in Swagger
guards	List[Callable]	No	[]	Applied to every route in this controller

---

@get, @post, @put, @delete, @patch

Route decorators. Applied to methods inside a @controller class.

Parameters (all share the same signature):

Parameter	Type	Required	Default	Description
path	str	No	""	Route path appended to the controller prefix
summary	str	No	None	Short one-line description shown in Swagger
description	str	No	None	Longer description shown in Swagger
guards	List[Callable]	No	[]	Merged with controller guards
**kwargs	—	No	—	Passed to the underlying FastAPI route decorator (e.g. response_model, status_code)

---

@websocket

Decorator. Marks a class as a WebSocket handler.

Parameters:

Parameter	Type	Required	Default	Description
path	str	Yes	—	WebSocket URL path
description	str	No	""	Shown in /stream-docs
guards	List[Callable]	No	[]	Run at connect time — async guard(websocket)

---

@on_message

Decorator. Marks a method inside a @websocket class as a message handler.

Parameters:

Parameter	Type	Required	Default	Description
description	str	No	""	Shown in /stream-docs
args	dict	No	{}	{arg_name: description} shown in /stream-docs
guards	List[Callable]	No	[]	Run per-message — async guard(websocket, data)

---

@sse

Decorator. Marks a class as an SSE handler.

Parameters:

Parameter	Type	Required	Default	Description
path	str	Yes	—	SSE endpoint URL path
description	str	No	""	Shown in /stream-docs
guards	List[Callable]	No	[]	Run before the stream starts

---

@sse_stream

Decorator. Marks a method inside an @sse class as the event stream generator.

Parameters:

Parameter	Type	Required	Default	Description
description	str	No	""	Shown in /stream-docs
args	dict	No	{}	{arg_name: description} shown in /stream-docs
guards	List[Callable]	No	[]	Run before the stream starts

---

VsAPIException

Base exception class for all server-side HTTP error responses. Extends VsBaseException from vs-common. All subclasses are handled automatically by VsFastApiServer.

Constructor:

Parameter	Type	Description
status_code	int	HTTP status code
error_code	str	Machine-readable error code
message	str	Human-readable error message
details	Optional[Dict]	Additional error context

Subclasses:

Class	Status	Code
NotFoundException	404	NOT_FOUND
ValidationException	400	VALIDATION_ERROR
UnauthorizedException	401	UNAUTHORIZED
ForbiddenException	403	FORBIDDEN
InternalServerException	500	INTERNAL_ERROR
ServiceUnavailableException	503	SERVICE_UNAVAILABLE
RateLimitException	429	RATE_LIMIT_EXCEEDED
TimeoutException	504	TIMEOUT_ERROR

RateLimitException accepts an additional retry_after: int parameter (seconds) and sets the Retry-After response header.

---

VsServerHealth

Pydantic model returned by GET /health.

Fields:

Field	Type	Description
status	str	healthy, degraded, or unhealthy
name	str	Server name
version	str	Server version
uptime_seconds	Optional[float]	Seconds since startup
cache	Optional[str]	healthy or unhealthy
instance_id	Optional[str]	Hostname of the running instance
dependencies	Dict[str, str]	Custom dependency statuses from _check_dependencies()
timestamp	datetime	UTC timestamp of the health check

---

VsDocsRenderer

Generates the /stream-docs HTML page for WebSocket and SSE endpoints.

Constructor:

Parameter	Type	Description
name	str	Server name shown in the page header
version	str	Server version shown in the page header

Methods:

Method	Signature	Description
render	render(ws_endpoints: list, sse_endpoints: list) -> str	Returns a complete HTML string for the docs page