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

Library	Required	Purpose
vs-common	Yes	Config, logging
pydantic	Yes	Request/response schema validation
fastapi	No — install with [fastapi] extra	FastAPI HTTP server implementation
uvicorn	No — install with [fastapi] extra	ASGI server
starlette	No — pulled in by FastAPI	CORS middleware
vs-db	No	Database session middleware (register VsDbMiddleware manually)
vs-security	No	JWT auth for guards

---

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, /vs-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 /vs-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

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, /vs-docs

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():
    ...

---

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, guards=[], **kwargs)	GET
@post(path, guards=[], **kwargs)	POST
@put(path, guards=[], **kwargs)	PUT
@delete(path, guards=[], **kwargs)	DELETE
@patch(path, 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 /vs-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 /vs-docs
args	dict	No	{}	Argument descriptions shown in /vs-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.

@sse parameters

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

@sse_stream parameters

Parameter	Type	Required	Default	Description
description	str	No	""	Shown in /vs-docs
args	dict	No	{}	Argument descriptions shown in /vs-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 /vs-docs	Custom HTML docs page for WebSocket and SSE endpoints — hidden when server.docs_enabled = false

If server.base_url is set in config.ini, /, /health, and /vs-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)
/vs-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 /vs-docs instead, generated by VsDocsRenderer.

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

---

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

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. Catch by base type to handle errors uniformly.

Automatically handled Python exceptions:

Exception	HTTP Status
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()	Reads server.reload	Force hot-reload on/off
get_workers()	Reads server.workers	Set worker count programmatically
get_log_level()	Reads server.log_level	Set log level programmatically
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, GET /vs-docs.

Methods (beyond VsServer contract):

Method	Description
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
guards	List[Callable]	No	[]	Merged with controller guards
**kwargs	—	No	—	Passed to the underlying FastAPI route decorator (e.g. response_model, status_code, summary)

---

@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 /vs-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 /vs-docs
args	dict	No	{}	{arg_name: description} shown in /vs-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 /vs-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 /vs-docs
args	dict	No	{}	{arg_name: description} shown in /vs-docs
guards	List[Callable]	No	[]	Run before the stream starts

---

VsAPIException

Base exception class for all HTTP error responses. 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 /vs-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