Embedding in Rust

pgvis is not just a standalone server — it’s a library you can embed in any Rust application. The pgvis-lib crate provides a Builder that assembles the full stack in one call.

Add the Dependency

[dependencies]
pgvis-lib = "0.1"
tokio = { version = "1", features = ["full"] }
axum = "0.8"
serde_json = "1"
anyhow = "1"

The pgvis-lib crate includes all backends and MCP by default. Feature flags:

Feature	Default	Description
postgres	✓	PostgreSQL backend
sqlite	✓	SQLite backend
mcp	✓	MCP server support

pgvis-lib re-exports the crates it builds on — pgvis_lib::pgvis_core, pgvis_lib::pgvis_router, and (per feature) pgvis_lib::pgvis_mcp, pgvis_lib::pgvis_postgres, pgvis_lib::pgvis_sqlite — so you don’t need to depend on them directly.

Quick Start: REST Router

use pgvis_lib::Builder;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let router = Builder::new("postgres://localhost/mydb")
        .schemas(vec!["public"])
        .build()
        .await?;

    let listener = tokio::net::TcpListener::bind("127.0.0.1:3000").await?;
    axum::serve(listener, router).await?;
    Ok(())
}

With no jwt_secret and no anon_role configured, every request runs as the DSN’s own role — see Authentication before exposing the router.

REST + MCP (Streamable HTTP)

use pgvis_lib::Builder;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let router = Builder::new("postgres://localhost/mydb")
        .schemas(vec!["public", "analytics"])
        .with_mcp_http()
        .build()
        .await?;

    // Serves REST at /api/{schema}/{table} AND MCP at /mcp
    let listener = tokio::net::TcpListener::bind("127.0.0.1:3000").await?;
    axum::serve(listener, router).await?;
    Ok(())
}

MCP-only Server (stdio)

For Claude Desktop or other agent integrations:

use pgvis_lib::Builder;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let mcp = Builder::new("postgres://localhost/mydb")
        .schemas(vec!["public"])
        .build_mcp_server()
        .await?;

    pgvis_lib::pgvis_mcp::serve_stdio(mcp)
        .await
        .map_err(|e| anyhow::anyhow!(e))?;
    Ok(())
}

MCP tool calls always run as anon_role — see the MCP guide.

SQLite Backend

The backend is auto-detected from the DSN:

use pgvis_lib::Builder;

// File-based SQLite
let router = Builder::new("sqlite:./mydata.db")
    .build()
    .await?;

// In-memory SQLite (useful for testing)
let router = Builder::new("sqlite::memory:")
    .build()
    .await?;

SQLite has a single main schema, which is the default for SQLite DSNs. Leave .schemas() unset: SQLite introspection ignores it, but the names you pass are still used for routing. A full Config passed with .config(...) defaults schemas to ["public"], so set schemas: vec!["main".into()] there.

Custom Configuration

use pgvis_lib::Builder;
use pgvis_lib::pgvis_core::config::{Config, RoutingConfig, JwtAlgorithm};

let config = Config {
    schemas: vec!["public".into(), "api".into()],
    max_rows: Some(1000),
    statement_timeout_ms: Some(10_000),
    jwt_secret: Some("my-secret-key".into()),
    jwt_algo: JwtAlgorithm::HS256,
    anon_role: Some("web_anon".into()),
    aggregates_enabled: true,
    read_only: false,
    routing: RoutingConfig {
        prefix: "v1".into(),
        schema_in_path: true,
        ..Default::default()
    },
    ..Default::default()
};

let router = Builder::new("postgres://localhost/mydb")
    .config(config)
    .build()
    .await?;

Authentication

The router authenticates HTTP requests from the Config you pass in:

Configuration	Request without a token	Request with a valid JWT
No jwt_secret	Runs as anon_role, or as the DSN’s role if unset	Token is not checked; same as without
jwt_secret + anon_role	Runs as anon_role	Runs as the token’s role claim (role_claim_key), or anon_role if it has none
jwt_secret, no anon_role	401	Runs as the token’s role claim; a token without one is 401 rather than falling back to the DSN’s role

jwt_aud (when set) rejects tokens minted for another audience, and nbf is honoured. The verified claims are available to SQL as one GUC, request.jwt.claims — pgvis no longer sets per-claim request.jwt.claim.<key> settings:

current_setting('request.jwt.claims', true)::json->>'sub'

The pgvis serve binary refuses to start with neither jwt_secret nor anon_role unless --insecure-no-auth is passed, and rejects unknown config-file keys. The library does neither: it is up to you to configure auth, or to put your own authentication in front of the router and call pgvis in-process with an explicit CallerIdentity.

Accessing Internal Components

Use build_components() for access to the schema cache, backend, and other internals:

use pgvis_lib::Builder;

let components = Builder::new("postgres://localhost/mydb")
    .schemas(vec!["public"])
    .build_components()
    .await?;

// Access the schema cache
let cache = components.cache.load();
println!("Tables: {}", cache.tables.len());
for table in cache.tables.values() {
    println!("  {}.{} ({} columns)",
        table.schema(), table.name(), table.columns.len());
}

// Generate OpenAPI spec
let spec = pgvis_lib::pgvis_router::openapi::generate_spec(&cache, &components.config);
let json = serde_json::to_string_pretty(&spec)?;
println!("{json}");

// Serve the router
let listener = tokio::net::TcpListener::bind("127.0.0.1:3000").await?;
axum::serve(listener, components.router).await?;

Components struct

Field	Type	Description
backend	Arc<dyn Backend>	Database backend (execute queries)
cache	Arc<ArcSwap<SchemaCache>>	Hot-swappable schema cache
config	Arc<Config>	Shared configuration
dialect	Arc<Dialect>	SQL dialect (Postgres/SQLite)
router	axum::Router	The fully-wired router: REST + OpenAPI, plus /mcp and /pubsub when enabled
pubsub	Option<Arc<PubSubHub>>	The pub/sub hub; None when pub/sub is disabled or the backend is SQLite
reloader	SchemaReloader	Keeps cache current after DDL (see below)

Components is #[non_exhaustive]: read or move its fields, but don’t construct or exhaustively destructure it.

Keeping the Schema Cache Current

The Builder spawns a SchemaReloader that re-introspects the database and atomically swaps in the new schema cache — REST routes, OpenAPI and MCP tools all follow it. In-flight requests finish on the cache they started with, and a failed introspection keeps the current cache.

On Postgres it reloads on PostgREST’s NOTIFY pgrst, 'reload schema' (or an empty payload), received on a dedicated LISTEN connection. Send it from your migrations, or install a DDL event trigger:

CREATE OR REPLACE FUNCTION pgrst_watch() RETURNS event_trigger
  LANGUAGE plpgsql AS $$ BEGIN NOTIFY pgrst, 'reload schema'; END; $$;
CREATE EVENT TRIGGER pgrst_watch ON ddl_command_end EXECUTE PROCEDURE pgrst_watch();

You can also reload on demand through components.reloader:

// Fire and forget; requests arriving together are coalesced into one
// introspection after a 200 ms debounce.
components.reloader.reload();

// Or reload now and wait for it — e.g. right after running migrations.
run_migrations().await?;
components.reloader.reload_now().await?; // new tables are routable from here on

SchemaReloader is Clone, so you can hand it to a signal handler or admin endpoint (pgvis serve reloads on SIGUSR1). If you build your own cache instead of using the Builder, start one with pgvis_lib::SchemaReloader::spawn(backend, cache, introspect_config). When the data cache is enabled, a schema reload also invalidates every cached response.

Calling Functions (RPC) In-Process

Beyond serving REST endpoints, pgvis lets you call database functions directly from Rust — no HTTP round-trip required. This is the programmatic equivalent of POST /rpc/{function}, but runs in-process through the same pipeline (plan → render → execute) with full Row-Level Security support.

Setup: Shared AppState

To use in-process RPC alongside HTTP serving, construct an AppState once and share it between the router and your application code:

use pgvis_lib::Builder;
use pgvis_lib::pgvis_router::{AppState, CallerIdentity, build_router};

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let components = Builder::new("postgres://localhost/mydb")
        .schemas(vec!["public"])
        .build_components()
        .await?;

    // Build shared state — used by both the router and in-process calls
    let state = AppState::new(
        components.cache,
        components.config,
        components.dialect,
        components.backend,
    );

    // Wire the HTTP router from the shared state
    let router = build_router(state.clone());

    // Spawn the HTTP server
    let listener = tokio::net::TcpListener::bind("127.0.0.1:3000").await?;
    tokio::spawn(async move { axum::serve(listener, router).await.ok(); });

    // Call a function in-process (same pipeline, no HTTP)
    let result = state
        .call_rpc("public", "add", serde_json::json!({"a": 3, "b": 5}), &CallerIdentity::anonymous())
        .await?;
    println!("3 + 5 = {}", result.body); // 3 + 5 = [{"result":8}]

    Ok(())
}

build_router(state) vs build_app(...) — Use build_router when you need to keep a reference to the AppState for in-process RPC. build_app is a convenience that constructs the state internally and returns only the router. AppState::new creates its own data cache (when config.cache.enabled), so serve the router built from that state rather than components.router — then HTTP and in-process calls share one cache, and a mutating in-process call invalidates what HTTP readers see. build_router serves REST and OpenAPI only; mount /mcp and /pubsub yourself if you need them.

call_rpc — Raw Result

let result = state.call_rpc(
    "public",           // schema
    "search_items",     // function name
    serde_json::json!({"query": "widget"}), // named arguments
    &CallerIdentity::anonymous(),
).await?;

// result.body is a serde_json::Value: an array of rows for set-returning
// functions, `[{"result": <value>}]` for scalar ones
let items: Vec<serde_json::Value> = serde_json::from_value(result.body)?;
println!("Found {} items", items.len());

Overloaded functions are resolved by argument names, as on the HTTP path: an unknown or misspelled argument, or a missing required one, is PGRST202; arguments matching more than one overload are PGRST203.

Typed Helpers

#[derive(serde::Deserialize)]
struct Item {
    id: i64,
    name: String,
    price: f64,
}

// Set-returning function (RETURNS TABLE / SETOF) → one T per row
let items: Vec<Item> = state
    .call_rpc_rows("public", "get_items", serde_json::json!({}), &CallerIdentity::anonymous())
    .await?;

// Scalar (or single JSON object) function → the unwrapped value
let sum: i64 = state
    .call_rpc_scalar("public", "add", serde_json::json!({"a": 10, "b": 20}), &CallerIdentity::anonymous())
    .await?;
assert_eq!(sum, 30);

// Nullable scalar → None for SQL NULL or no row
let maybe: Option<String> = state
    .call_rpc_scalar_opt("public", "find_name", serde_json::json!({"id": 7}), &CallerIdentity::anonymous())
    .await?;

call_read — In-Process Table Reads

Reads use the same PostgREST query syntax as GET /{table}, and always bypass the data cache:

use std::collections::HashMap;

let params = HashMap::from([
    ("select".to_string(), "id,name,price".to_string()),
    ("price".to_string(), "lt.10".to_string()),
    ("order".to_string(), "name.asc".to_string()),
    ("limit".to_string(), "20".to_string()),
]);

let cheap: Vec<Item> = state
    .call_read_rows("public", "items", &params, &CallerIdentity::with_role("authenticated"))
    .await?;

CallerIdentity — Row-Level Security

CallerIdentity controls the database role and JWT claims applied inside the transaction, exactly as if the call came from an authenticated HTTP request. There is no JWT verification on this path — your application is vouching for the identity:

use pgvis_lib::pgvis_router::CallerIdentity;

// Anonymous — no role switch, no claims
let anon = CallerIdentity::anonymous();

// Role-only — SET LOCAL role = 'authenticated'
let user = CallerIdentity::with_role("authenticated");

// Full identity — role + JWT claims propagated as the request.jwt.claims GUC
let admin = CallerIdentity {
    role: Some("admin".into()),
    claims: Some(serde_json::json!({
        "sub": "user-123",
        "role": "admin",
        "org_id": "org-456",
    })),
};

// RLS policies using current_setting('request.jwt.claims', true)::json->>'org_id' see these claims
let result = state
    .call_rpc("public", "get_my_data", serde_json::json!({}), &admin)
    .await?;

CallerIdentity::anonymous() does not apply anon_role: with no role it runs as the DSN’s own role. Pass CallerIdentity::with_role(...) with your anonymous role when the call should be restricted.

API Reference

Method	Signature	Returns
call_rpc	(&self, schema, function, args, caller) → Result<QueryResult, Error>	Raw result with body, total_count, page_total
call_rpc_rows<T>	(&self, schema, function, args, caller) → Result<Vec<T>, Error>	Each row of a set-returning function as T
call_rpc_scalar<T>	(&self, schema, function, args, caller) → Result<T, Error>	The single value of a scalar function; error on NULL / no row
call_rpc_scalar_opt<T>	(&self, schema, function, args, caller) → Result<Option<T>, Error>	As above, None on NULL / no row
call_read	(&self, schema, table, &params, caller) → Result<QueryResult, Error>	Raw table read; params is a HashMap<String, String> of query parameters
call_read_rows<T>	(&self, schema, table, &params, caller) → Result<Vec<T>, Error>	Each row as T

CallerIdentity	Description
::anonymous()	No role switch (runs as the DSN’s role), no claims
::with_role("name")	SET LOCAL role, no claims
{ role, claims }	Full identity with GUC propagation

Error Handling

call_rpc returns pgvis_core::error::Error:

use pgvis_lib::pgvis_core::error::{Error, ErrorCode};

match state.call_rpc("public", "nonexistent_fn", serde_json::json!({}), &CallerIdentity::anonymous()).await {
    Ok(result) => { /* use result.body */ },
    Err(Error::Plan { code: ErrorCode::FunctionNotFound, message, .. }) => eprintln!("Function not found: {message}"),
    Err(Error::Execution { message, db_code, .. }) => eprintln!("DB error {db_code:?}: {message}"),
    Err(e) => eprintln!("Other error ({}): {e}", e.code().as_str()),
}

---

Composing with Your Own Routes

Since pgvis returns a standard axum::Router, you can compose it with your own routes:

use axum::{Router, routing::get};
use pgvis_lib::Builder;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let pgvis_router = Builder::new("postgres://localhost/mydb")
        .build()
        .await?;

    let app = Router::new()
        .route("/health", get(|| async { "ok" }))
        .route("/custom", get(|| async { "my custom endpoint" }))
        .merge(pgvis_router);

    let listener = tokio::net::TcpListener::bind("127.0.0.1:3000").await?;
    axum::serve(listener, app).await?;
    Ok(())
}

pgvis does not yet ship health or Prometheus metrics endpoints (they are designed but not implemented); the /health route above is your own.

Graceful Shutdown

Serving is up to you, so is shutdown. Use axum’s graceful shutdown to let in-flight requests finish, as pgvis serve does on SIGTERM / Ctrl-C:

axum::serve(listener, app)
    .with_graceful_shutdown(async {
        let _ = tokio::signal::ctrl_c().await;
    })
    .await?;

A request dropped mid-flight never hands an open transaction back to the pool: its connection is discarded and the server rolls it back.

Read Replicas and Pool Status

Set config.replica.replica_dsns or opt into config.replica.discovery and the Builder uses a replica-aware Postgres backend. Metadata-eligible base table reads and STABLE/IMMUTABLE RPCs can use readers; writes, VOLATILE RPCs, views, RLS, computed relationships and pre-request hooks stay primary. Reader transactions enforce READ ONLY. See discovery and consistency limits.

use pgvis_lib::pgvis_core::config::{Config, ReplicaConfig};

let config = Config {
    replica: ReplicaConfig {
        replica_dsns: vec!["postgres://replica1/mydb".into(), "postgres://replica2/mydb".into()],
        max_replication_lag_bytes: 10 * 1024 * 1024, // exclude replicas further behind
        health_check_interval_ms: 5_000,
        primary_reads: true,                         // primary also serves reads
        ..Default::default()
    },
    ..Default::default()
};

A background health monitor probes every replica concurrently over its own connections, under a deadline, and excludes unreachable or lagging ones; if none are eligible, reads fall back to the primary.

Backend::pool_status() reports the request pool’s occupancy for your own monitoring (None for SQLite; the primary pool for the replica backend):

if let Some(pool) = components.backend.pool_status() {
    println!("{}/{} connections, {} idle, {} waiting",
        pool.size, pool.max_size, pool.available, pool.waiting);
}

When the pool is exhausted, requests fail with 503 (PGRST000) rather than 500.

Data Cache

Set config.cache.enabled = true to cache read responses in memory (primary-key lookups, plus list queries with cache_lists = true) for ttl_seconds, up to max_entries. REST and in-process mutations invalidate the affected table, volatile RPCs and schema reloads invalidate everything. MCP tool calls are not cached, but MCP writes also don’t invalidate the REST data cache yet — keep TTLs short if agents write to tables that HTTP clients read through the cache.

Error Handling

Builder::build() and build_components() return Result<_, pgvis_core::error::Error>. Common failure cases:

• Database connection failed (wrong DSN, server unreachable)
• Introspection failed (insufficient permissions)

match Builder::new("postgres://localhost/mydb").build().await {
    Ok(router) => { /* serve */ },
    Err(e) => {
        eprintln!("pgvis initialization failed: {e}");
        std::process::exit(1);
    }
}

Architecture

The embed API (pgvis-lib) is the same code path the CLI uses. There is no “server-only” logic. build_components() picks the backend from the DSN, introspects once, then wires every surface to the same backend and schema cache (dashed parts are optional: /pubsub needs pubsub.enabled on Postgres, /mcp needs with_mcp_http()):

flowchart TB
    b(["Builder::new(dsn)"]) --> kind{"detect_db_kind"}
    kind -->|Postgres| pg["<b>PgBackend</b><br/>or PgReplicaBackend"]
    kind -->|SQLite| sq["<b>SqliteBackend</b>"]
    pg --> intro["<b>introspect</b><br/>SchemaCache"]
    sq --> intro
    intro --> rel["<b>SchemaReloader</b>"]
    intro --> app["<b>build_app</b><br/>REST + OpenAPI"]
    intro -.-> ps["<b>/pubsub</b><br/>if enabled"]
    intro -.-> mcp["<b>/mcp</b><br/>with_mcp_http"]
    rel --> comp(["Components"])
    app --> comp
    ps -.-> comp
    mcp -.-> comp

In words, the Builder assembles:

1. A database backend (connection pool + introspection; replica-aware when replicas are configured)
2. A schema cache (tables, columns, relationships, routines, constraints) and its SchemaReloader
3. A dialect (Postgres or SQLite capability flags)
4. Configuration
5. An axum Router (REST + OpenAPI) or MCP Server

All wired together with the same I/O-free core engine. The pipeline is:

ApiRequest → plan_request() → ActionPlan → render() → SQL → execute() → QueryResult

This pipeline is shared between REST dispatch and MCP tool execution — behaviour is identical regardless of surface.

Pub/Sub (Real-time Messaging)

When pub/sub is enabled in your config, the Components struct exposes a pubsub field — an Arc<PubSubHub> you can use directly:

use pgvis_lib::Builder;
use pgvis_lib::pgvis_core::pubsub::PubSubConfig;
use pgvis_lib::pgvis_core::config::Config;

let config = Config {
    pubsub: PubSubConfig { enabled: true, ..Default::default() },
    ..Default::default()
};

let components = Builder::new("postgres://localhost/mydb")
    .config(config)
    .build_components()
    .await?;

if let Some(hub) = &components.pubsub {
    // Publish
    hub.publish("events", r#"{"type":"user_signup"}"#).await?;

    // Subscribe
    let mut rx = hub.subscribe("events").await?;
    tokio::spawn(async move {
        while let Ok(msg) = rx.recv().await {
            println!("{}: {}", msg.channel, msg.payload);
        }
    });
}

The hub is trusted in-process code: it publishes as the DSN’s role and skips pubsub.authorize_function, which applies only to the REST and MCP surfaces. The REST router already includes /pubsub/* endpoints automatically when enabled. See the full Pub/Sub guide for configuration, authorization, patterns, and limitations.