FUSE Stack

from sqlmodel import SQLModel, Field
from typing import List

class UserBase(SQLModel):
    username: str = Field(min_length=3, max_length=32, unique=True, index=True)
    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
    email: str = Field(unique=True, index=True)

class UserInDB(UserBase, table=True):
    password_hash: str = Field(max_length=255) # We DON'T want to send this sensitive information to the frontend.
    # So, we store it only in the DB model.

class UserPublic(UserBase):
    ...
    # We just define this as an empty class, because there is nothing not in UserBase we want to send to the frontend.


# A complex response containing nested models.
class TeamResponse(BaseModel):
    team_name: str
    members: List[UserPublic]

@router.get('/team')
async def get_team() -> TeamResponse:
    # Fetch raw database objects
    db_user_1 = UserInDB(id=1, username="alice", password_hash="hash1")
    db_user_2 = UserInDB(id=2, username="bob", password_hash="hash2")

    # We pass the DB objects directly into the response model.
    # Pydantic will automatically strip the 'password_hash' field 
    # because it doesn't exist in 'UserPublic'!
    return TeamResponse(
        team_name="Engineering",
        members=[db_user_1, db_user_2]
    )
    # So, we can sleep easy knowing that our password hashes aren't leaking!

// `team` is automatically type-hinted as a TeamResponse!
const team = await client.get_team();

console.log(team.team_name); // "Engineering"

// The members list is fully typed.
team.members.forEach(member => {
    console.log(member.username); 
    // console.log(member.password_hash); // TypeScript Error! Field does not exist.
    // It won't even error at runtime, it will literally show the error in your editor before you even finish writing :D
});

const res = await fetch('/api/team');
const team = await res.json();

if (!res.ok) throw new Error(team.error || "Unknown error.");

// NO autocompletion here! You need to guess the types instead :(
console.log(team.team_name);

import asyncio
from typing import AsyncGenerator
from openai import AsyncOpenAI

client = AsyncOpenAI()

@router.get('/chat/stream')
async def stream_ai(prompt: str) -> AsyncGenerator[str, None]: # Synchronous generators work too!
    stream = await client.chat.completions.create(
        model="gpt-5",
        messages=[{"role": "user", "content": prompt}],
        stream=True,
    )

    async for chunk in stream:
        if chunk.choices[0].delta.content is not None:
            # Just yield the string. FUSE handles the rest.
            yield chunk.choices[0].delta.content

// We get a stream object back instantly
const stream = await client.stream_ai("What is the capital of France?");

// Normal async iteration, that you already know how to do.
for await (const token of stream) {
    // `token` is type-hinted as a string because of the Python return signature!
    document.getElementById('chat').innerHTML += token;
}

// The HTTP fallback is powerful, but requires manual parsing.
const res = await fetch('/api/chat/stream?prompt=Write a poem about FUSE.');
// Another issue with using HTTP instead of Ephaptic is that you'd need to manually `encodeURIComponent` the query params (we don't do this now for brevity).

const reader = res.body.getReader();
const decoder = new TextDecoder();

while (true) {
    const { value, done } = await reader.read();
    if (done) break;

    // FUSE streams pure JSONL for HTTP, so we just split by newline and parse
    const lines = decoder.decode(value).split('\n');
    for (const line of lines) {
        if (line.trim()) {
            const token = JSON.parse(line); // Manual parsing
            // Again, you get no type hinting.
            document.getElementById('chat').innerHTML += token;
        }
    }
}

// HTTP has significantly more boilerplate. But it does work.

from pydantic import BaseModel

class JobStatusEvent(BaseModel):
    job_id: str
    status: str
    progress: int

class JobStartResponse(BaseModel):
    started: bool

@router.post('/jobs/start')
async def start_job(job_id: str) -> JobStartResponse:
    # We start the job, but we can emit events to specific users instantly
    await ephaptic.emit( # Shorthand for `ephaptic.to(active_user()).emit()`
        JobStatusEvent(job_id=job_id, status="Processing", progress=50)
    )
    return {"started": True} # You are allowed to return dicts here! You don't always *need* to instantiate a model, as long as types match.

// TypeScript knows exactly what `event` looks like.
client.on('JobStatusEvent', (event) => {
    console.log(`Job ${event.job_id} is at ${event.progress}%`);
    document.querySelector('progress').value = event.progress;
});

// Start the job
const { started } = await client.start_job("job_123");
// Using JS object destructuring! and TypeScript will *still* know that `started` is a bool thanks to the power of Ephaptic!

from fastapi import Request
import os
import jwt
from datetime import datetime, timezone, timedelta

JWT_SECRET = os.getenv('JWT_SECRET')

def generate_token(user_id):
    return jwt.encode({
        'sub': str(user_id),
        'exp': datetime.now(timezone.utc) + timedelta(weeks=52)
    }, JWT_SECRET, algorithm='HS256')

def verify_token(token):
    try:
        payload = jwt.decode(token, JWT_SECRET, algorithms=['HS256'])
        return payload['sub']
    except: return None

@ephaptic.identity_loader
def load_identity(auth):
    if not auth: return None
    token = auth.get('token')
    return verify_token(token) if token else None

@ephaptic.http_identity_loader
def load_identity_http(request: Request):
    token = request.headers.get('Authorization') or request.cookies.get('Authorization')
    return verify_token(token.removeprefix('Bearer ')) if token else None

from ephaptic.ctx import active_user, is_http, is_rpc

class Client(BaseModel):
    user_id: Optional[str] # Can be None if not logged in, because identity loader can return None!
    am_i_ephaptic_client: bool
    am_i_http_client: bool # Either one or the other will always be True.

# Now we can define our special function!
@router.get('/whoami', limit='5/h') # This limit is imposed both via WebSocket AND via HTTP using a client's IP address if not logged in, or the client's user ID when logged in!
async def whoAmI() -> Client:
    return Client(
        user_id = active_user(),

        am_i_ephaptic_client = is_rpc(),
        am_i_http_client     = is_http(),
    )

import { connect } from "@ephaptic/client";

const client = connect();

const me = await client.whoAmI();

console.log(me.user_id);              // null (not logged in)
console.log(me.am_i_ephaptic_client); // true
console.log(me.am_i_http_client);     // false

import { connect } from "@ephaptic/client";

const client = connect({ auth: { token: window.localStorage.getItem('token') } });

const me = await client.whoAmI();

console.log(me.user_id);              // some string matching the JWT 
console.log(me.am_i_ephaptic_client); // true
console.log(me.am_i_http_client);     // false

const token = window.localStorage.getItem('token');

const res = await fetch('/api/whoami', {
    headers: {
        'Authorization': `Bearer ${token}`
    }
});
const me = await res.json();

// `me` is not type hinted.
console.log(me.user_id);              // some string matching the JWT 
console.log(me.am_i_ephaptic_client); // false
console.log(me.am_i_http_client);     // true