Repository: modiimedia/arri (language: rust)

Issue:
websocket transport support



Discussion / hints:

Fix patch:
diff --git a/packages/arri-client-dart/lib/arri_client.dart b/packages/arri-client-dart/lib/arri_client.dart
index 3e4c8e5d..846ae860 100644
--- a/packages/arri-client-dart/lib/arri_client.dart
+++ b/packages/arri-client-dart/lib/arri_client.dart
@@ -4,3 +4,4 @@ export "errors.dart";
 export 'parsing.dart';
 export 'request.dart';
 export 'sse.dart';
+export 'ws.dart';
diff --git a/packages/arri-client-dart/lib/errors.dart b/packages/arri-client-dart/lib/errors.dart
index cef22227..bae3dbdf 100644
--- a/packages/arri-client-dart/lib/errors.dart
+++ b/packages/arri-client-dart/lib/errors.dart
@@ -64,6 +64,20 @@ class ArriError implements Exception {
     );
   }
 
+  Map<String, dynamic> toJson() {
+    final result = <String, dynamic>{
+      "code": code,
+      "message": message,
+    };
+    if (data != null) {
+      result["data"] = data;
+    }
+    if (stack != null) {
+      result["stack"] = stack;
+    }
+    return result;
+  }
+
   factory ArriError.fromString(String input) {
     try {
       final val = json.decode(input);
diff --git a/packages/arri-client-dart/lib/ws.dart b/packages/arri-client-dart/lib/ws.dart
new file mode 100644
index 00000000..3fa62646
--- /dev/null
+++ b/packages/arri-client-dart/lib/ws.dart
@@ -0,0 +1,136 @@
+import 'dart:async';
+
+import 'package:arri_client/arri_client.dart';
+import 'package:web_socket_channel/web_socket_channel.dart';
+import 'package:web_socket_channel/status.dart' as status;
+
+Future<ArriWebsocketController<TServerMessage, TClientMessage>>
+    arriWebsocketRequest<TServerMessage, TClientMessage>(
+  String url, {
+  Map<String, dynamic>? headers,
+  required TServerMessage Function(String msg) parser,
+  required String Function(TClientMessage msg) serializer,
+}) async {
+  var finalUrl =
+      url.replaceAll("http://", "ws://").replaceAll("https://", "wss://");
+  if (headers != null) {
+    final queryParts = <String>[];
+    for (final entry in headers.entries) {
+      queryParts.add("${entry.key}=${entry.value}");
+    }
+    finalUrl = "$finalUrl?${queryParts.join("&")}";
+  }
+  final uri = Uri.parse(finalUrl);
+  final channel = WebSocketChannel.connect(uri);
+  await channel.ready;
+  return ArriWebsocketController<TServerMessage, TClientMessage>(
+    channel: channel,
+    parser: parser,
+    serializer: serializer,
+  );
+}
+
+typedef WsMessageHandler<T> = Function(T data);
+typedef MessageParser<T> = T Function(String data);
+typedef MessageSerializer<T> = String Function(T data);
+
+class ArriWebsocketController<TIncoming, TOutgoing> {
+  final WebSocketChannel channel;
+  final MessageParser<TIncoming> parser;
+  final MessageSerializer<TOutgoing> serializer;
+
+  bool isClosed = false;
+  ArriWebsocketController({
+    required this.channel,
+    required this.parser,
+    required this.serializer,
+  });
+
+  void send(TOutgoing msg) {
+    channel.sink.add(serializer(msg));
+  }
+
+  void close([int? statusCode = status.goingAway, String? reason]) {
+    channel.sink.close(statusCode, reason);
+  }
+
+  StreamSubscription listen({
+    required Function(TIncoming data) onData,
+    Function(WsEvent<TIncoming> event)? onEvent,
+    Function(ArriError error)? onError,
+    Function(Object?)? onConnectionError,
+    Function()? onDone,
+    bool? cancelOnError,
+  }) {
+    return channel.stream.listen(
+      (message) {
+        if (message is! String) {
+          return;
+        }
+        final event = WsEvent.fromString(message, parser);
+        onEvent?.call(event);
+        switch (event) {
+          case WsMessageEvent<TIncoming>():
+            onData(event.data);
+            break;
+          case WsErrorEvent<TIncoming>():
+            onError?.call(event.data);
+            break;
+          case WsRawEvent<TIncoming>():
+            break;
+        }
+      },
+      onError: onConnectionError,
+      onDone: onDone,
+      cancelOnError: cancelOnError,
+    );
+  }
+}
+
+/// implemented by [WsMessageEvent], [WsErrorEvent], and [WsRawEvent]
+sealed class WsEvent<T> {
+  factory WsEvent.fromString(String input, T Function(String data) parser) {
+    final event = WsRawEvent<T>.fromString(input);
+    switch (event.type) {
+      case "message":
+        return WsMessageEvent(parser(event.data));
+      case "error":
+        return WsErrorEvent(ArriError.fromString(event.data));
+      default:
+        return event;
+    }
+  }
+}
+
+class WsMessageEvent<T> implements WsEvent<T> {
+  final T data;
+  const WsMessageEvent(this.data);
+}
+
+class WsErrorEvent<T> implements WsEvent<T> {
+  final ArriError data;
+  const WsErrorEvent(this.data);
+}
+
+class WsRawEvent<T> implements WsEvent<T> {
+  final String type;
+  final String data;
+  const WsRawEvent({required this.type, required this.data});
+
+  factory WsRawEvent.fromString(String input) {
+    final lines = input.split("\n");
+    String type = "unknown";
+    String data = "";
+    for (final line in lines) {
+      if (line.startsWith("type:")) {
+        type = line.replaceFirst("type:", "").trim();
+        continue;
+      }
+      if (line.startsWith("data:")) {
+        data = line.replaceFirst("data:", "");
+        continue;
+      }
+    }
+    return WsRawEvent(type: type, data: data);
+  }
+}
diff --git a/packages/arri-client-dart/pubspec.yaml b/packages/arri-client-dart/pubspec.yaml
index 3996ea13..12d7c831 100644
--- a/packages/arri-client-dart/pubspec.yaml
+++ b/packages/arri-client-dart/pubspec.yaml
@@ -7,6 +7,7 @@ environment:
   sdk: ">=3.0.0 <4.0.0"
 dependencies:
   http: ">=0.13.0 <2.0.0"
+  web_socket_channel: ">=2.4.4 <3.0.0"
 dev_dependencies:
   lints: ^3.0.0
   test: ^1.25.1
diff --git a/packages/arri-client/package.json b/packages/arri-client/package.json
index eecd7b16..955aac37 100644
--- a/packages/arri-client/package.json
+++ b/packages/arri-client/package.json
@@ -17,6 +17,10 @@
   "dependencies": {
     "@fortaine/fetch-event-source": "^3.0.6",
     "arri-validate": "workspace:*",
-    "ofetch": "^1.3.4"
+    "ofetch": "^1.3.4",
+    "ws": "^8.16.0"
+  },
+  "devDependencies": {
+    "@types/ws": "^8.5.10"
   }
 }
diff --git a/packages/arri-client/src/_index.ts b/packages/arri-client/src/_index.ts
index f139f90b..e93fb457 100644
--- a/packages/arri-client/src/_index.ts
+++ b/packages/arri-client/src/_index.ts
@@ -1,3 +1,4 @@
 export * from "./errors";
 export * from "./request";
 export * from "./sse";
+export * from "./ws";
diff --git a/packages/arri-client/src/ws.ts b/packages/arri-client/src/ws.ts
new file mode 100644
index 00000000..b58169c7
--- /dev/null
+++ b/packages/arri-client/src/ws.ts
@@ -0,0 +1,209 @@
+import NodeWebsocket from "ws";
+import { type ArriError, ArriErrorInstance } from "./errors";
+
+function isBrowser() {
+    return typeof window !== "undefined";
+}
+
+interface WsControllerOptions<TParams, TResponse> {
+    url: string;
+    serializer: (input: TParams) => string;
+    parser: (input: unknown) => TResponse;
+    onMessage?: WsMessageHook<TResponse>;
+    onError?: WsErrorHook;
+    onConnectionError?: WsErrorHook;
+    onOpen?: () => any;
+    onClose?: () => any;
+}
+
+export interface WsOptions<TResponse> {
+    onMessage?: WsMessageHook<TResponse>;
+    onError?: WsErrorHook;
+    onConnectionError?: WsErrorHook;
+    onOpen?: () => any;
+    onClose?: () => any;
+}
+
+interface ArriWsRequestOptions<TParams = any, TResponse = any> {
+    url: string;
+    headers?: Record<string, string>;
+    params?: TParams;
+    parser: (input: unknown) => TResponse;
+    serializer: (input: TParams) => string;
+    onMessage?: WsMessageHook<TResponse>;
+    onError?: WsErrorHook;
+    onConnectionError?: WsErrorHook;
+    onOpen?: () => any;
+    onClose?: () => any;
+}
+
+function connectWebsocket(url: string, protocol?: string) {
+    if (isBrowser()) {
+        return new WebSocket(url, protocol);
+    }
+    return new NodeWebsocket(url, {
+        protocol,
+    });
+}
+
+export function arriWsRequest<
+    TParams extends Record<any, any> | undefined = undefined,
+    TResponse = any,
+>(
+    opts: ArriWsRequestOptions<TParams, TResponse>,
+    retryCount = 0,
+): WsController<TParams, TResponse> {
+    let url = opts.url
+        .replace("http://", "ws://")
+        .replace("https://", "wss://");
+    if (opts.headers) {
+        const queryParts: string[] = [];
+        for (const key of Object.keys(opts.headers)) {
+            queryParts.push(`${key}=${opts.headers[key]}`);
+        }
+        url += `?${queryParts.join("&")}`;
+    }
+    try {
+        const controller = new WsController<TParams, TResponse>({
+            url,
+            parser: opts.parser,
+            serializer: opts.serializer ?? ((_) => ""),
+            onOpen: opts.onOpen,
+            onClose: opts.onClose,
+            onMessage: opts.onMessage,
+            onError: opts.onError,
+            onConnectionError: opts.onConnectionError,
+        });
+        return controller;
+    } catch (err) {
+        console.error(err);
+        if (opts.onConnectionError) {
+            // eslint-disable-next-line @typescript-eslint/no-unsafe-argument
+            opts.onConnectionError(err as any);
+        }
+        return arriWsRequest(opts, retryCount + 1);
+    }
+}
+
+type WsErrorHook = (err: ArriError) => void;
+type WsMessageHook<TResponse> = (msg: TResponse) => any;
+
+class WsController<TParams, TResponse> {
+    url: string;
+    private _ws?: NodeWebsocket | WebSocket;
+    private readonly _serializer: (input: TParams) => string;
+    private readonly _parser: (input: unknown) => TResponse;
+    onMessage?: WsMessageHook<TResponse>;
+    onError?: WsErrorHook;
+    onConnectionError?: WsErrorHook;
+    onOpen?: () => any;
+    onClose?: () => any;
+    constructor(opts: WsControllerOptions<TParams, TResponse>) {
+        this.url = opts.url;
+        this._serializer = opts.serializer;
+        this._parser = opts.parser;
+        this.onOpen = opts.onOpen;
+        this.onClose = opts.onClose;
+        this.onError = opts.onError;
+        this.onConnectionError = opts.onConnectionError;
+        this.onMessage = opts.onMessage;
+    }
+
+    connect() {
+        this._ws = connectWebsocket(this.url);
+        this._ws.onopen = () => {
+            if (this.onOpen) {
+                this.onOpen();
+            }
+        };
+        this._ws.onclose = () => {
+            if (this.onClose) {
+                this.onClose();
+            }
+        };
+        this._ws.onmessage = (event: MessageEvent) => {
+            this._handleMessage(event);
+        };
+        this._ws.onerror = (event: any) => {
+            console.error(event);
+            if (this.onConnectionError) {
+                this.onConnectionError({
+                    code: 500,
+                    message: "IDK",
+                });
+            }
+        };
+    }
+
+    readyState() {
+        return this._ws?.readyState ?? -1;
+    }
+
+    send(message: TParams) {
+        const payload = this._serializer(message);
+        this._ws?.send(payload);
+    }
+
+    close() {
+        this._ws?.close();
+    }
+
+    private _handleMessage(msg: MessageEvent) {
+        if (typeof msg.data !== "string") {
+            return;
+        }
+        const response = parsedWsResponse(msg.data);
+        switch (response.type) {
+            case "error": {
+                if (!this.onError) {
+                    return;
+                }
+                const err = ArriErrorInstance.fromJson(response.data);
+                this.onError(err);
+                break;
+            }
+            case "message": {
+                if (!this.onMessage) {
+                    return;
+                }
+                const msg = this._parser(response.data);
+                this.onMessage(msg);
+                break;
+            }
+            default:
+                console.warn("Invalid response from server", msg);
+                break;
+        }
+    }
+}
+
+function parsedWsResponse(input: string): {
+    type: "message" | "error" | "unknown";
+    data: string;
+} {
+    const lines = input.split("\n");
+    let type: "message" | "error" | "unknown" = "unknown";
+    let data = "";
+    for (const line of lines) {
+        const trimmedLine = line.trim();
+        if (trimmedLine.startsWith("type: ")) {
+            switch (trimmedLine.substring(5).trim()) {
+                case "error":
+                    type = "error";
+                    break;
+                case "message":
+                    type = "message";
+                    break;
+            }
+            continue;
+        }
+        if (trimmedLine.startsWith("data: ")) {
+            data = trimmedLine.substring(5).trim();
+            continue;
+        }
+    }
+    return {
+        type,
+        data,
+    };
+}
diff --git a/packages/arri-codegen/dart-reference/project.json b/packages/arri-codegen/dart-reference/project.json
index 95df8fa7..43ebe8c6 100644
--- a/packages/arri-codegen/dart-reference/project.json
+++ b/packages/arri-codegen/dart-reference/project.json
@@ -2,5 +2,14 @@
   "name": "arri-codegen-dart-reference",
   "$schema": "../../../node_modules/nx/schemas/project-schema.json",
   "type": "library",
-  "sourceRoot": "packages/arri-codegen/dart-reference"
+  "sourceRoot": "packages/arri-codegen/dart-reference",
+  "targets": {
+    "pub": {
+      "executor": "nx:run-commands",
+      "options": {
+        "command": "dart pub",
+        "cwd": "packages/arri-codegen/dart-reference"
+      }
+    }
+  }
 }
diff --git a/packages/arri-codegen/dart-reference/pubspec.lock b/packages/arri-codegen/dart-reference/pubspec.lock
index 52a909cb..e47e19a5 100644
--- a/packages/arri-codegen/dart-reference/pubspec.lock
+++ b/packages/arri-codegen/dart-reference/pubspec.lock
@@ -31,7 +31,7 @@ packages:
       path: "../../arri-client-dart"
       relative: true
     source: path
-    version: "0.36.0"
+    version: "0.36.1"
   async:
     dependency: transitive
     description:
@@ -148,10 +148,10 @@ packages:
     dependency: "direct dev"
     description:
       name: lints
-      sha256: "0a217c6c989d21039f1498c3ed9f3ed71b354e69873f13a8dfc3c9fe76f1b452"
+      sha256: cbf8d4b858bb0134ef3ef87841abdf8d63bfc255c266b7bf6b39daa1085c4290
       url: "https://pub.dev"
     source: hosted
-    version: "2.1.1"
+    version: "3.0.0"
   logging:
     dependency: transitive
     description:
diff --git a/packages/arri-codegen/dart-reference/pubspec.yaml b/packages/arri-codegen/dart-reference/pubspec.yaml
index 09566c7b..52ab82d0 100644
--- a/packages/arri-codegen/dart-reference/pubspec.yaml
+++ b/packages/arri-codegen/dart-reference/pubspec.yaml
@@ -9,5 +9,5 @@ dependencies:
   arri_client:
     path: "../../arri-client-dart"
 dev_dependencies:
-  lints: ^2.0.0
+  lints: ^3.0.0
   test: ^1.21.0
diff --git a/packages/arri-codegen/dart-reference/reference_client.dart b/packages/arri-codegen/dart-reference/reference_client.dart
index a4ebcbbf..d23d7064 100644
--- a/packages/arri-codegen/dart-reference/reference_client.dart
+++ b/packages/arri-codegen/dart-reference/reference_client.dart
@@ -115,6 +115,15 @@ class TestClientUsersService {
       lastEventId: lastEventId,
     );
   }
+
+  Future<ArriWebsocketController<User, UserParams>> createConnection() {
+    return arriWebsocketRequest<User, UserParams>(
+      "$_baseUrl/users/create-connection",
+      headers: _headers,
+      parser: (body) => User.fromJson(json.decode(body)),
+      serializer: (body) => json.encode(body.toJson()),
+    );
+  }
 }
 
 class TestClientUsersSettingsService {
diff --git a/packages/arri-codegen/dart/src/index.ts b/packages/arri-codegen/dart/src/index.ts
index 1c998c3e..09cd6bb5 100644
--- a/packages/arri-codegen/dart/src/index.ts
+++ b/packages/arri-codegen/dart/src/index.ts
@@ -26,6 +26,8 @@ import {
     type SchemaMetadata,
     isSchemaFormRef,
     type SchemaFormRef,
+    type WsRpcDefinition,
+    type HttpRpcDefinition,
 } from "arri-codegen-utils";
 import { a } from "arri-validate";
 
@@ -252,12 +254,23 @@ export function dartRpcFromDefinition(
     def: RpcDefinition,
     opts: DartClientGeneratorOptions,
 ): string {
-    if (def.transport !== "http") {
-        console.warn(
-            `[codegen-dart] WARNING: non-http RPCs not supported at this time`,
-        );
-        return "";
+    if (def.transport === "http") {
+        return dartHttpRpcFromSchema(key, def, opts);
     }
+    if (def.transport === "ws") {
+        return dartWsRpcFromSchema(key, def, opts);
+    }
+    console.warn(
+        `[codegen-dart] WARNING: unsupported transport "${def.transport}". Skipping "${def.path}".`,
+    );
+    return "";
+}
+
+export function dartHttpRpcFromSchema(
+    key: string,
+    def: HttpRpcDefinition,
+    opts: DartClientGeneratorOptions,
+): string {
     let returnType:
         | `Future<String>`
         | "Future<int>"
@@ -354,6 +367,34 @@ export function dartRpcFromDefinition(
   }`;
 }
 
+export function dartWsRpcFromSchema(
+    key: string,
+    def: WsRpcDefinition,
+    opts: DartClientGeneratorOptions,
+): string {
+    const serverMsg = def.response
+        ? pascalCase(def.response, { normalize: true })
+        : "null";
+    const clientMsg = def.params
+        ? pascalCase(def.params, { normalize: true })
+        : "null";
+    const returnType = `Future<ArriWebsocketController<${serverMsg}, ${clientMsg}>>`;
+    const parser = def.response
+        ? `(body) => ${serverMsg}.fromJson(json.decode(body))`
+        : `(_) => null`;
+    const serializer = def.params
+        ? `(body) => json.encode(body.toJson())`
+        : `(_) => {}`;
+    return `${getAnnotations({ description: def.description, isDeprecated: def.isDeprecated })}${returnType} ${key}() {
+        return arriWebsocketRequest<${serverMsg}, ${clientMsg}>(
+            "$_baseUrl${def.path}",
+            headers: _headers,
+            parser: ${parser},
+            serializer: ${serializer},
+        );
+    }`;
+}
+
 export function dartTypeFromJtdSchema(
     /**
      * location in the tree i.e User.reviews.id
diff --git a/packages/arri-codegen/typescript/package.json b/packages/arri-codegen/typescript/package.json
index 0d7c7c4b..deae473f 100644
--- a/packages/arri-codegen/typescript/package.json
+++ b/packages/arri-codegen/typescript/package.json
@@ -1,9 +1,15 @@
 {
   "name": "arri-codegen-ts",
+  "version": "0.36.1",
   "type": "module",
   "license": "MIT",
-  "author": { "name": "joshmossas", "url": "https://github.com/joshmossas" },
-  "bugs": { "url": "https://github.com/modiimedia/arri/issues" },
+  "author": {
+    "name": "joshmossas",
+    "url": "https://github.com/joshmossas"
+  },
+  "bugs": {
+    "url": "https://github.com/modiimedia/arri/issues"
+  },
   "repository": {
     "type": "git",
     "url": "https://github.com/modiimedia/arri.git",
@@ -12,13 +18,16 @@
   "main": "./dist/index.cjs",
   "module": "./dist/index.mjs",
   "types": "./dist/index.d.ts",
-  "files": ["dist"],
+  "files": [
+    "dist"
+  ],
   "dependencies": {
     "ajv": "^8.12.0",
-    "arri-client": "workspace:*",
     "arri-codegen-utils": "workspace:*",
     "arri-validate": "workspace:*",
     "prettier": "^3.2.5"
   },
-  "version": "0.36.1"
+  "devDependencies": {
+    "arri-client": "workspace:*"
+  }
 }
diff --git a/packages/arri-codegen/typescript/references/wsExample.ts b/packages/arri-codegen/typescript/references/wsExample.ts
new file mode 100644
index 00000000..3d460320
--- /dev/null
+++ b/packages/arri-codegen/typescript/references/wsExample.ts
@@ -0,0 +1,59 @@
+/* eslint-disable @typescript-eslint/unbound-method */
+import { arriWsRequest, type WsOptions } from "arri-client";
+
+export class UserService {
+    private readonly baseUrl: string;
+    private readonly headers: Record<string, string>;
+
+    constructor(opts: { baseUrl?: string; headers?: Record<string, string> }) {
+        this.baseUrl = opts.baseUrl ?? "";
+        this.headers = opts.headers ?? {};
+    }
+
+    createConnection(opts: WsOptions<ServerMessage>) {
+        return arriWsRequest<ClientMessage, ServerMessage>({
+            url: `${this.baseUrl}/users/create-connection`,
+            headers: this.headers,
+            parser: $$ClientMessage.parse,
+            serializer: $$ClientMessage.serialize,
+            onOpen: opts.onOpen,
+            onClose: opts.onClose,
+            onError: opts.onError,
+            onConnectionError: opts.onConnectionError,
+            onMessage: opts.onMessage,
+        });
+    }
+}
+
+export interface ServerMessage {
+    id: string;
+    content: string;
+}
+export const $$ServerMessage = {
+    parse(input: unknown): ServerMessage {
+        return {
+            id: "",
+            content: "",
+        };
+    },
+    serialize(input: ServerMessage): string {
+        return "";
+    },
+};
+
+export interface ClientMessage {
+    id: string;
+    content: string;
+}
+
+export const $$ClientMessage = {
+    parse(input: unknown): ClientMessage {
+        return {
+            id: "",
+            content: "",
+        };
+    },
+    serialize(input: ClientMessage): string {
+        return "";
+    },
+};
diff --git a/packages/arri-codegen/typescript/src/index.ts b/packages/arri-codegen/typescript/src/index.ts
index ff3576b0..c6030af3 100644
--- a/packages/arri-codegen/typescript/src/index.ts
+++ b/packages/arri-codegen/typescript/src/index.ts
@@ -22,6 +22,8 @@ import {
     isRpcDefinition,
     isServiceDefinition,
     type ServiceDefinition,
+    type HttpRpcDefinition,
+    type WsRpcDefinition,
     isSchemaFormRef,
     type SchemaFormRef,
 } from "arri-codegen-utils";
@@ -74,6 +76,7 @@ export async function createTypescriptClient(
         versionNumber: def.info?.version ?? "",
         typesNeedingParser: Object.keys(def.models),
         hasSseProcedures: false,
+        hasWsProcedures: false,
     };
     Object.keys(services).forEach((key) => {
         const item = services[key];
@@ -117,6 +120,10 @@ export async function createTypescriptClient(
         importParts.push("arriSseRequest");
         importParts.push("type SseOptions");
     }
+    if (rpcOptions.hasWsProcedures) {
+        importParts.push("arriWsRequest");
+        importParts.push("type WsOptions");
+    }
 
     // generated only types if no procedures are found
     if (procedureParts.length === 0 && serviceFieldParts.length === 0) {
@@ -167,6 +174,7 @@ interface RpcOptions extends GeneratorOptions {
     versionNumber: string;
     typesNeedingParser: string[];
     hasSseProcedures: boolean;
+    hasWsProcedures: boolean;
 }
 
 export function tsRpcFromDefinition(
@@ -174,12 +182,23 @@ export function tsRpcFromDefinition(
     schema: RpcDefinition,
     options: RpcOptions,
 ): string {
-    if (schema.transport !== "http") {
-        console.warn(
-            "[codegen-ts] WARNING: Non-http RPCs are not supported at this time",
-        );
-        return "";
+    if (schema.transport === "http") {
+        return tsHttpRpcFromDefinition(key, schema, options);
+    }
+    if (schema.transport === "ws") {
+        return tsWsRpcFromDefinition(key, schema, options);
     }
+    console.warn(
+        `[codegen-ts] Unsupported transport "${schema.transport}". Ignoring rpc.`,
+    );
+    return "";
+}
+
+export function tsHttpRpcFromDefinition(
+    key: string,
+    schema: HttpRpcDefinition,
+    options: RpcOptions,
+) {
     const paramName = pascalCase(schema.params ?? "");
     const responseName = pascalCase(schema.response ?? "");
     const paramsInput = schema.params ? `params: ${paramName}` : "";
@@ -224,6 +243,38 @@ export function tsRpcFromDefinition(
         });
     }`;
 }
+export function tsWsRpcFromDefinition(
+    key: string,
+    schema: WsRpcDefinition,
+    options: RpcOptions,
+): string {
+    options.hasWsProcedures = true;
+    const paramName = pascalCase(schema.params ?? "undefined");
+    const responseName = pascalCase(schema.response ?? "undefined");
+    const hasInput = paramName.length > 0;
+    const hasOutput = responseName.length > 0;
+    let serializerPart = `(_) => {}`;
+    let parserPart = `(_) => {}`;
+    if (hasInput) {
+        serializerPart = `$$${paramName}.serialize`;
+    }
+    if (hasOutput) {
+        parserPart = `$$${responseName}.parse`;
+    }
+    return `${getJsDocComment(schema)}${key}(options: WsOptions<${responseName}> = {}) {
+        return arriWsRequest<${paramName}, ${responseName}>({
+            url: \`\${this.baseUrl}${schema.path}\`,
+            headers: this.headers,
+            parser: ${parserPart},
+            serializer: ${serializerPart},
+            onOpen: options.onOpen,
+            onClose: options.onClose,
+            onError: options.onError,
+            onConnectionError: options.onConnectionError,
+            onMessage: options.onMessage,
+        })
+    }`;
+}
 
 export function tsServiceFromDefinition(
     name: string,
diff --git a/packages/arri-validate/src/lib/validation.ts b/packages/arri-validate/src/lib/validation.ts
index 1b5363c9..2b607012 100644
--- a/packages/arri-validate/src/lib/validation.ts
+++ b/packages/arri-validate/src/lib/validation.ts
@@ -135,11 +135,16 @@ export function serialize<T = any>(schema: ASchema<T>, input: T) {
 
 export function errors(schema: ASchema, input: unknown): ValueError[] {
     const errorList: ValueError[] = [];
-    schema.metadata[SCHEMA_METADATA].parse(input, {
-        errors: errorList,
-        instancePath: "",
-        schemaPath: "",
-    });
+    try {
+        schema.metadata[SCHEMA_METADATA].parse(input, {
+            errors: errorList,
+            instancePath: "",
+            schemaPath: "",
+        });
+    } catch (err) {
+        // eslint-disable-next-line @typescript-eslint/restrict-template-expressions
+        errorList.push({ instancePath: "", schemaPath: "", message: `${err}` });
+    }
     return errorList;
 }
 
diff --git a/packages/arri/package.json b/packages/arri/package.json
index 84f986f7..6ce82131 100644
--- a/packages/arri/package.json
+++ b/packages/arri/package.json
@@ -25,6 +25,7 @@
     "chokidar": "^3.6.0",
     "citty": "^0.1.6",
     "consola": "^3.2.3",
+    "crossws": "^0.2.4",
     "degit": "^2.8.4",
     "enquirer": "^2.4.1",
     "esbuild": "^0.20.2",
diff --git a/packages/arri/src/_index.ts b/packages/arri/src/_index.ts
index b8a255ec..a50d0186 100644
--- a/packages/arri/src/_index.ts
+++ b/packages/arri/src/_index.ts
@@ -6,4 +6,5 @@ export * from "./config";
 export * from "./router";
 export * from "./route";
 export * from "./eventStreamRpc";
+export * from "./websocketRpc";
 export * from "h3";
diff --git a/packages/arri/src/app.ts b/packages/arri/src/app.ts
index eae5d586..eb7c7194 100644
--- a/packages/arri/src/app.ts
+++ b/packages/arri/src/app.ts
@@ -29,7 +29,13 @@ import {
     registerRpc,
     isRpcParamSchema,
     type NamedRpc,
+    type RpcParamSchema,
 } from "./rpc";
+import {
+    createWsRpcDefinition,
+    registerWebsocketRpc,
+    type NamedWebsocketRpc,
+} from "./websocketRpc";
 
 export const DEV_ENDPOINT_ROOT = `/__arri_dev__`;
 export const DEV_DEFINITION_ENDPOINT = `${DEV_ENDPOINT_ROOT}/__definition`;
@@ -117,7 +123,11 @@ export class ArriApp implements ArriRouterBase {
                 this.route(route);
             }
             for (const rpc of input.getProcedures()) {
-                this.rpc(rpc);
+                if (rpc.transport === "http") {
+                    this.rpc(rpc);
+                } else {
+                    this.wsRpc(rpc);
+                }
             }
             this.registerModels(input.getModels());
             return;
@@ -129,28 +139,33 @@ export class ArriApp implements ArriRouterBase {
         TIsEventStream extends boolean = false,
         TParams extends AObjectSchema<any, any> | undefined = undefined,
         TResponse extends AObjectSchema<any, any> | undefined = undefined,
-    >(procedure: NamedRpc<TIsEventStream, TParams, TResponse>) {
-        const path =
-            procedure.path ?? getRpcPath(procedure.name, this._rpcRoutePrefix);
-        this._procedures[procedure.name] = createHttpRpcDefinition(
-            procedure.name,
-            path,
-            procedure,
-        );
-        if (isRpcParamSchema(procedure.params)) {
-            const paramName = getRpcParamName(procedure.name, procedure);
+    >(
+        procedure: Omit<
+            NamedRpc<TIsEventStream, TParams, TResponse>,
+            "transport"
+        >,
+    ) {
+        (procedure as any).transport = "http";
+        const p = procedure as NamedRpc<TIsEventStream, TParams, TResponse>;
+        const path = p.path ?? getRpcPath(p.name, this._rpcRoutePrefix);
+        this._procedures[p.name] = createHttpRpcDefinition(p.name, path, p);
+
+        if (isRpcParamSchema(p.params)) {
+            // eslint-disable-next-line @typescript-eslint/no-unsafe-argument
+            const paramName = getRpcParamName(p.name, p);
             if (paramName) {
-                this._models[paramName] = procedure.params;
+                this._models[paramName] = p.params;
             }
         }
-        if (isRpcParamSchema(procedure.response)) {
-            const responseName = getRpcResponseName(procedure.name, procedure);
+        if (isRpcParamSchema(p.response)) {
+            // eslint-disable-next-line @typescript-eslint/no-unsafe-argument
+            const responseName = getRpcResponseName(p.name, p as any);
             if (responseName) {
-                this._models[responseName] = procedure.response;
+                this._models[responseName] = p.response;
             }
         }
-        if (isEventStreamRpc(procedure)) {
-            registerEventStreamRpc(this.h3Router, path, procedure, {
+        if (isEventStreamRpc(p)) {
+            registerEventStreamRpc(this.h3Router, path, p, {
                 middleware: this._middlewares,
                 onRequest: this._onRequest,
                 onError: this._onError,
@@ -160,7 +175,7 @@ export class ArriApp implements ArriRouterBase {
             });
             return;
         }
-        registerRpc(this.h3Router, path, procedure, {
+        registerRpc(this.h3Router, path, p, {
             middleware: this._middlewares,
             onRequest: this._onRequest,
             onError: this._onError,
@@ -170,6 +185,34 @@ export class ArriApp implements ArriRouterBase {
         });
     }
 
+    wsRpc<
+        TParams extends RpcParamSchema | undefined,
+        TResponse extends RpcParamSchema | undefined,
+    >(procedure: Omit<NamedWebsocketRpc<TParams, TResponse>, "transport">) {
+        (procedure as any).transport = "ws";
+        const p = procedure as NamedWebsocketRpc<TParams, TResponse>;
+        const path =
+            procedure.path ?? getRpcPath(procedure.name, this._rpcRoutePrefix);
+        this._procedures[procedure.name] = createWsRpcDefinition(
+            procedure.name,
+            path,
+            p,
+        );
+        if (isRpcParamSchema(procedure.params)) {
+            const paramName = getRpcParamName(procedure.name, p);
+            if (paramName) {
+                this._models[paramName] = procedure.params;
+            }
+        }
+        if (isRpcParamSchema(procedure.response)) {
+            const responseName = getRpcResponseName(procedure.name, p);
+            if (responseName) {
+                this._models[responseName] = procedure.response;
+            }
+        }
+        registerWebsocketRpc(this.h3Router, path, p);
+    }
+
     route<
         TPath extends string,
         TQuery extends AObjectSchema<any, any>,
diff --git a/packages/arri/src/cli/_common.ts b/packages/arri/src/cli/_common.ts
index c3dc107a..0deb7df1 100644
--- a/packages/arri/src/cli/_common.ts
+++ b/packages/arri/src/cli/_common.ts
@@ -77,10 +77,17 @@ export async function createAppWithRoutesModule(config: ResolvedArriConfig) {
             .join(",\n")}
         ];
         for(const route of routes) {
-            app.rpc({
-                name: route.id,
-                ...route.route,
-            });
+            if(route.route.transport === 'http') {
+                app.rpc({
+                    name: route.id,
+                    ...route.route,
+                });
+            } else {
+                app.wsRpc({
+                    name: route.id,
+                    ...route.route,
+                });
+            }
         }
         export default app`,
         { parser: "typescript", tabWidth: 4 },
diff --git a/packages/arri/src/cli/build.ts b/packages/arri/src/cli/build.ts
index 52f4d9ee..bd1e399d 100644
--- a/packages/arri/src/cli/build.ts
+++ b/packages/arri/src/cli/build.ts
@@ -171,6 +171,14 @@ import app from './${OUT_APP_FILE}';
 void listen(toNodeListener(app.h3App), {
     port: process.env.PORT ?? ${config.port},
     public: true,
+    ws: {
+        resolve(info) {
+            if (app.h3App.websocket?.resolve) {
+                return app.h3App.websocket.resolve(info);
+            }
+            return app.h3App.websocket?.hooks ?? app.h3App.handler?.__websocket__ ?? {};
+        }
+    },
     http2: ${config.http2 ?? false},
     ${httpsString}
 });`;
diff --git a/packages/arri/src/cli/dev.ts b/packages/arri/src/cli/dev.ts
index a841e6cd..9d231338 100644
--- a/packages/arri/src/cli/dev.ts
+++ b/packages/arri/src/cli/dev.ts
@@ -62,6 +62,7 @@ const startListener = (config: ResolvedArriConfig, showQr = false) =>
         qr: showQr,
         https: config.https,
         http2: config.http2,
+        ws: true,
     });
 
 async function bundleFilesContext(config: ResolvedArriConfig) {
diff --git a/packages/arri/src/eventStreamRpc.ts b/packages/arri/src/eventStreamRpc.ts
index 12ef2e6d..494d0ea8 100644
--- a/packages/arri/src/eventStreamRpc.ts
+++ b/packages/arri/src/eventStreamRpc.ts
@@ -32,12 +32,16 @@ export function defineEventStreamRpc<
     TParams extends RpcParamSchema | undefined = undefined,
     TResponse extends RpcParamSchema | undefined | never = undefined,
 >(
-    config: Omit<EventStreamRpc<TParams, TResponse>, "isEventStream">,
+    config: Omit<
+        EventStreamRpc<TParams, TResponse>,
+        "isEventStream" | "transport"
+    >,
 ): EventStreamRpc<TParams, TResponse> {
     return {
         ...config,
         method: config.method ?? "get",
         isEventStream: true,
+        transport: "http",
     };
 }
 
diff --git a/packages/arri/src/router.ts b/packages/arri/src/router.ts
index 7217f2dc..c9e6fe5e 100644
--- a/packages/arri/src/router.ts
+++ b/packages/arri/src/router.ts
@@ -1,7 +1,8 @@
 import { type ASchema, type AObjectSchema } from "arri-validate";
 import { type ModelMap } from "./app";
 import { type ArriRoute } from "./route";
-import { type NamedRpc } from "./rpc";
+import { type RpcParamSchema, type NamedRpc } from "./rpc";
+import { type NamedWebsocketRpc } from "./websocketRpc";
 
 export interface ArriRouterBase {
     rpc: <
@@ -9,7 +10,16 @@ export interface ArriRouterBase {
         TParams extends AObjectSchema<any, any> | undefined,
         TResponse extends AObjectSchema<any, any> | undefined,
     >(
-        procedure: NamedRpc<TIsEventStream, TParams, TResponse>,
+        procedure: Omit<
+            NamedRpc<TIsEventStream, TParams, TResponse>,
+            "transport"
+        >,
+    ) => void;
+    wsRpc: <
+        TParams extends RpcParamSchema | undefined,
+        TResponse extends RpcParamSchema | undefined,
+    >(
+        procedure: Omit<NamedWebsocketRpc<TParams, TResponse>, "transport">,
     ) => void;
     route: <
         TPath extends string,
@@ -24,7 +34,9 @@ export interface ArriRouterBase {
 }
 
 export class ArriRouter implements ArriRouterBase {
-    private readonly procedures: Array<NamedRpc<any, any, any>> = [];
+    private readonly procedures: Array<
+        NamedRpc<any, any, any> | NamedWebsocketRpc<any, any>
+    > = [];
 
     private readonly routes: Array<ArriRoute<any>> = [];
 
@@ -34,8 +46,24 @@ export class ArriRouter implements ArriRouterBase {
         TIsEventStream extends boolean = false,
         TParams extends AObjectSchema<any, any> | undefined = undefined,
         TResponse extends AObjectSchema<any, any> | undefined = undefined,
-    >(procedure: NamedRpc<TIsEventStream, TParams, TResponse>) {
-        this.procedures.push(procedure);
+    >(
+        procedure: Omit<
+            NamedRpc<TIsEventStream, TParams, TResponse>,
+            "transport"
+        >,
+    ) {
+        (procedure as any).transport = "http";
+        // eslint-disable-next-line @typescript-eslint/no-unsafe-argument
+        this.procedures.push(procedure as any);
+    }
+
+    wsRpc<
+        TParams extends RpcParamSchema | undefined,
+        TResponse extends RpcParamSchema | undefined,
+    >(procedure: Omit<NamedWebsocketRpc<TParams, TResponse>, "transport">) {
+        (procedure as any).transport = "ws";
+        // eslint-disable-next-line @typescript-eslint/no-unsafe-argument
+        this.procedures.push(procedure as any);
     }
 
     route<
diff --git a/packages/arri/src/rpc.ts b/packages/arri/src/rpc.ts
index 29471847..d5e2b8d7 100644
--- a/packages/arri/src/rpc.ts
+++ b/packages/arri/src/rpc.ts
@@ -33,6 +33,7 @@ import {
 } from "./eventStreamRpc";
 import { type MiddlewareEvent } from "./middleware";
 import { type RouteOptions } from "./route";
+import { type WebsocketRpc } from "./websocketRpc";
 
 export type RpcParamSchema<TObjectInner = any, TDiscriminatorInner = any> =
     | AObjectSchema<TObjectInner>
@@ -55,6 +56,7 @@ export interface Rpc<
     TParams extends RpcParamSchema | undefined = undefined,
     TResponse extends RpcParamSchema | undefined = undefined,
 > {
+    transport: "http";
     method?: RpcHttpMethod;
     path?: string;
     description?: string;
@@ -139,8 +141,11 @@ export function isRpc(input: unknown): input is Rpc<any, any> {
 export function defineRpc<
     TParams extends RpcParamSchema | undefined = undefined,
     TResponse extends RpcParamSchema | undefined | never = undefined,
->(config: HttpRpc<TParams, TResponse>): Rpc<false, TParams, TResponse> {
-    return config;
+>(
+    config: Omit<HttpRpc<TParams, TResponse>, "transport">,
+): Rpc<false, TParams, TResponse> {
+    (config as any).transport = "http";
+    return config as any;
 }
 
 export function createHttpRpcDefinition(
@@ -182,7 +187,7 @@ export function getRpcPath(rpcName: string, prefix = ""): string {
 
 export function getRpcParamName(
     rpcName: string,
-    procedure: Rpc<any, any, any>,
+    procedure: Rpc<any, any, any> | WebsocketRpc<any, any>,
 ): string | undefined {
     if (!isRpcParamSchema(procedure.params)) {
         return undefined;
@@ -200,7 +205,7 @@ export function getRpcParamName(
 
 export function getRpcResponseName(
     rpcName: string,
-    procedure: Rpc<any, any, any>,
+    procedure: Rpc<any, any, any> | WebsocketRpc<any, any>,
 ): string | undefined {
     if (!isRpcParamSchema(procedure.response)) {
         return undefined;
@@ -218,7 +223,10 @@ export function getRpcResponseName(
 
 function getRpcResponseDefinition(
     rpcName: string,
-    procedure: Rpc<any, any> | EventStreamRpc<any, any>,
+    procedure:
+        | Rpc<any, any>
+        | EventStreamRpc<any, any>
+        | WebsocketRpc<any, any>,
 ): RpcDefinition["response"] {
     if (!isRpcParamSchema(procedure.response)) {
         return undefined;
diff --git a/packages/arri/src/websocketRpc.ts b/packages/arri/src/websocketRpc.ts
new file mode 100644
index 00000000..d7eb9eb1
--- /dev/null
+++ b/packages/arri/src/websocketRpc.ts
@@ -0,0 +1,217 @@
+import { type WsRpcDefinition } from "arri-codegen-utils";
+import {
+    type CompiledValidator,
+    type ASchema,
+    a,
+    type InferType,
+} from "arri-validate";
+import { type Peer, type WSError } from "crossws";
+import { defineWebSocketHandler, type Router } from "h3";
+import { type ArriServerErrorResponse } from "./errors";
+import {
+    getRpcParamName,
+    getRpcResponseName,
+    type RpcParamSchema,
+} from "./rpc";
+
+export interface NamedWebsocketRpc<
+    TParams extends RpcParamSchema | undefined = undefined,
+    TResponse extends RpcParamSchema | undefined = undefined,
+> extends WebsocketRpc<TParams, TResponse> {
+    name: string;
+}
+
+export interface WebsocketRpc<
+    TParams extends RpcParamSchema | undefined = undefined,
+    TResponse extends RpcParamSchema | undefined = undefined,
+> {
+    transport: "ws";
+    description?: string;
+    path?: string;
+    isDeprecated?: boolean;
+    params: TParams;
+    response: TResponse;
+    pingInterval?: boolean;
+    handler: WebSocketRpcHandler<
+        TParams extends RpcParamSchema ? InferType<TParams> : undefined,
+        TResponse extends RpcParamSchema ? InferType<TResponse> : undefined
+    >;
+}
+
+interface WsPeerOpts<TResponse> {
+    validator?: CompiledValidator<ASchema<TResponse>>;
+    context: WsPeerContext;
+}
+
+export interface WsPeerContext extends Record<string, any> {
+    rpcName: string;
+    queryParams?: Record<string, string>;
+    clientAddress?: string;
+}
+
+export class WsPeer<TResponse> {
+    private readonly _peer: Peer;
+    private readonly _validator?: CompiledValidator<ASchema<TResponse>>;
+
+    context: WsPeerContext;
+    url: string;
+
+    constructor(peer: Peer, opts: WsPeerOpts<TResponse>) {
+        this._peer = peer;
+        this._validator = opts.validator;
+        this.url = peer.url;
+        this.context = opts.context;
+    }
+
+    send(data: TResponse) {
+        if (!this._validator) {
+            return;
+        }
+        if (!this._validator.validate(data)) {
+            return;
+        }
+        const payload = this._validator.serialize(data);
+        return this._peer.send(`type: message\ndata: ${payload}`);
+    }
+
+    sendError(err: ArriServerErrorResponse) {
+        return this._peer.send(`type: error\ndata: ${JSON.stringify(err)}`);
+    }
+
+    close() {
+        throw new Error("close() is not yet implemented on WsPeer");
+    }
+
+    subscribe(channel: string) {
+        this._peer.subscribe(channel);
+    }
+
+    unsubscribe(channel: string) {
+        this._peer.unsubscribe(channel);
+    }
+
+    publish(channel: string, message: TResponse) {
+        if (!this._validator) {
+            return;
+        }
+        if (!this._validator.validate(message)) {
+            const err: ArriServerErrorResponse = {
+                code: 500,
+                message: `Error serializing message on server. The payload doesn't match the specified schema.`,
+                data: {
+                    payload: message,
+                },
+            };
+            this.sendError(err);
+            return;
+        }
+        const payload = this._validator.serialize(message);
+        this._peer.publish(channel, payload);
+    }
+
+    readyState() {
+        return this._peer.readyState;
+    }
+}
+
+export interface WebSocketRpcHandler<TParams, TResponse> {
+    onOpen: (peer: WsPeer<TResponse>) => void;
+    onMessage: (peer: WsPeer<TResponse>, message: TParams) => void;
+    onClose: (
+        peer: WsPeer<TResponse>,
+        details: { code?: number; reason?: string },
+    ) => void;
+    onError: (peer: WsPeer<TResponse>, error: WSError) => void;
+}
+
+export function defineWebsocketRpc<
+    TParams extends RpcParamSchema | undefined,
+    TResponse extends RpcParamSchema | undefined,
+>(
+    def: Omit<WebsocketRpc<TParams, TResponse>, "transport">,
+): WebsocketRpc<TParams, TResponse> {
+    (def as any).transport = "ws";
+    return def as any;
+}
+
+export function registerWebsocketRpc(
+    router: Router,
+    path: string,
+    rpc: NamedWebsocketRpc<any, any>,
+) {
+    let responseValidator: undefined | ReturnType<typeof a.compile>;
+    let paramValidator: undefined | ReturnType<typeof a.compile>;
+    try {
+        responseValidator = rpc.response ? a.compile(rpc.response) : undefined;
+    } catch (err) {
+        console.error("ERROR COMPILING VALIDATOR", err);
+    }
+    try {
+        paramValidator = rpc.params ? a.compile(rpc.params) : undefined;
+    } catch (err) {
+        console.error("ERROR COMPILING PARAMS", err);
+    }
+
+    const handler = defineWebSocketHandler({
+        upgrade(req) {},
+        open(peer) {
+            const urlParts = peer.url.split("?");
+            const context: WsPeerContext = {
+                rpcName: rpc.name,
+                clientAddress: peer.addr,
+            };
+            if (urlParts.length > 1) {
+                urlParts.shift();
+                const queryStr = new URLSearchParams(urlParts.join("?"));
+                const query: Record<string, string> = {};
+                for (const [key, val] of queryStr.entries()) {
+                    query[key] = val;
+                }
+                context.queryParams = query;
+            }
+            const wsPeer = new WsPeer(peer, {
+                validator: responseValidator,
+                context,
+            });
+            peer.ctx.__wsPeer = wsPeer;
+            rpc.handler.onOpen(peer.ctx.__wsPeer as WsPeer<any>);
+        },
+        message(peer, message) {
+            if (!paramValidator) {
+                return;
+            }
+            const data = paramValidator.safeParse(message.text());
+            if (!data.success) {
+                const errorResponse: ArriServerErrorResponse = {
+                    code: 400,
+                    message: data.error.message,
+                    data: data.error.errors,
+                    stack: data.error.stack?.split("\n"),
+                };
+                (peer.ctx.__wsPeer as WsPeer<any>).sendError(errorResponse);
+                return;
+            }
+            rpc.handler.onMessage(peer.ctx.__wsPeer as WsPeer<any>, data.value);
+        },
+        close(peer, details) {
+            rpc.handler.onClose(peer.ctx.__wsPeer as WsPeer<any>, details);
+        },
+    });
+
+    router.use(path, handler);
+}
+
+export function createWsRpcDefinition(
+    rpcName: string,
+    path: string,
+    rpc: NamedWebsocketRpc<any, any>,
+): WsRpcDefinition {
+    return {
+        transport: "ws",
+        path,
+        params: getRpcParamName(rpcName, rpc),
+        response: getRpcResponseName(rpcName, rpc),
+        isDeprecated: rpc.isDeprecated,
+        description: rpc.description,
+    };
+}
diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml
index a23ed31b..bcbb93d1 100644
--- a/pnpm-lock.yaml
+++ b/pnpm-lock.yaml
@@ -152,6 +152,9 @@ importers:
       consola:
         specifier: ^3.2.3
         version: 3.2.3
+      crossws:
+        specifier: ^0.2.4
+        version: 0.2.4
       degit:
         specifier: ^2.8.4
         version: 2.8.4
@@ -226,6 +229,13 @@ importers:
       ofetch:
         specifier: ^1.3.4
         version: 1.3.4
+      ws:
+        specifier: ^8.16.0
+        version: 8.16.0
+    devDependencies:
+      '@types/ws':
+        specifier: ^8.5.10
+        version: 8.5.10
 
   packages/arri-codegen/dart:
     dependencies:
@@ -250,9 +260,6 @@ importers:
       ajv:
         specifier: ^8.12.0
         version: 8.12.0
-      arri-client:
-        specifier: workspace:*
-        version: link:../../arri-client
       arri-codegen-utils:
         specifier: workspace:*
         version: link:../utils
@@ -262,6 +269,10 @@ importers:
       prettier:
         specifier: ^3.2.5
         version: 3.2.5
+    devDependencies:
+      arri-client:
+        specifier: workspace:*
+        version: link:../../arri-client
 
   packages/arri-codegen/utils:
     dependencies:
@@ -9510,6 +9521,19 @@ packages:
     resolution: {integrity: sha512-l4Sp/DRseor9wL6EvV2+TuQn63dMkPjZ/sp9XkghTEbV9KlPS1xUsZ3u7/IQO4wxtcFB4bgpQPRcR3QCvezPcQ==}
     dev: true
 
+  /ws@8.16.0:
+    resolution: {integrity: sha512-HS0c//TP7Ina87TfiPUz1rQzMhHrl/SG2guqRcTOIUYD2q8uhUdNHZYJUaQ8aTGPzCh+c6oawMKW35nFl1dxyQ==}
+    engines: {node: '>=10.0.0'}
+    peerDependencies:
+      bufferutil: ^4.0.1
+      utf-8-validate: '>=5.0.2'
+    peerDependenciesMeta:
+      bufferutil:
+        optional: true
+      utf-8-validate:
+        optional: true
+    dev: false
+
   /y18n@5.0.8:
     resolution: {integrity: sha512-0pfFzegeDWJHJIAmTLRP2DwHjdF5s7jo9tuztdQxAhINCdvS+3nGINqPd00AphqJR/0LhANUS6/+7SCb98YOfA==}
     engines: {node: '>=10'}
diff --git a/tools/scripts/publish.ts b/tools/scripts/publish.ts
index 50c2afbe..6c44358c 100644
--- a/tools/scripts/publish.ts
+++ b/tools/scripts/publish.ts
@@ -60,7 +60,7 @@ const main = defineCommand({
             }
         };
         const tasks = Object.keys(publishConfig).map(async (key) => {
-            const config = publishConfig[key];
+            const config = publishConfig[key]!;
             if (!config.allowPublish) {
                 return;
             }


Test patch:
diff --git a/packages/arri-codegen/typescript/src/index.test.ts b/packages/arri-codegen/typescript/src/index.test.ts
index 30c67c35..6ebda58d 100644
--- a/packages/arri-codegen/typescript/src/index.test.ts
+++ b/packages/arri-codegen/typescript/src/index.test.ts
@@ -51,6 +51,13 @@ describe("Service Creation", () => {
                 response: "User",
                 isEventStream: true,
             },
+            createConnection: {
+                transport: "ws",
+                description: "Create a ws connection to send messages",
+                path: "/create-connection",
+                params: "User",
+                response: "User",
+            },
         };
 
         const result = await prettier.format(
@@ -60,6 +67,7 @@ describe("Service Creation", () => {
                 typesNeedingParser: [],
                 versionNumber: "1",
                 hasSseProcedures: true,
+                hasWsProcedures: true,
             }),
             { parser: "typescript" },
         );
@@ -113,6 +121,22 @@ describe("Service Creation", () => {
                     serializer: $$GetUserParams.serialize,
                 }, options);
             }
+            /**
+             * Create a ws connection to send messages
+             */
+            createConnection(options: WsOptions<User> = {}) {
+                return arriWsRequest<User, User>({
+                    url: \`\${this.baseUrl}/create-connection\`,
+                    headers: this.headers,
+                    parser: $$User.parse,
+                    serializer: $$User.serialize,
+                    onOpen: options.onOpen,
+                    onClose: options.onClose,
+                    onError: options.onError,
+                    onConnectionError: options.onConnectionError,
+                    onMessage: options.onMessage,
+                });
+            }
         }`,
                     { parser: "typescript" },
                 ),
@@ -145,6 +169,7 @@ describe("Model Creation", () => {
                 versionNumber: "",
                 typesNeedingParser: ["User", "user"],
                 hasSseProcedures: false,
+                hasWsProcedures: false,
             },
             { existingTypeNames: [], isOptional: false },
         );
@@ -191,6 +216,7 @@ describe("Model Creation", () => {
                 versionNumber: "",
                 typesNeedingParser: ["User", "PartialUser"],
                 hasSseProcedures: false,
+                hasWsProcedures: false,
             },
             {
                 existingTypeNames: [],
diff --git a/tests/clients/client-dart/lib/test_client.rpc.dart b/tests/clients/client-dart/lib/test_client.rpc.dart
index 5acbd310..d15dab4a 100644
--- a/tests/clients/client-dart/lib/test_client.rpc.dart
+++ b/tests/clients/client-dart/lib/test_client.rpc.dart
@@ -337,6 +337,16 @@ class TestClientTestsService {
       lastEventId: lastEventId,
     );
   }
+
+  Future<ArriWebsocketController<WsMessageResponse, WsMessageParams>>
+      websocketRpc() {
+    return arriWebsocketRequest<WsMessageResponse, WsMessageParams>(
+      "$_baseUrl/rpcs/tests/websocket-rpc",
+      headers: _headers,
+      parser: (body) => WsMessageResponse.fromJson(json.decode(body)),
+      serializer: (body) => json.encode(body.toJson()),
+    );
+  }
 }
 
 class TestClientAdaptersService {
@@ -2820,3 +2830,256 @@ class ChatMessageUrl implements ChatMessage {
     );
   }
 }
+
+sealed class WsMessageParams {
+  final String type;
+  const WsMessageParams({
+    required this.type,
+  });
+  factory WsMessageParams.fromJson(Map<String, dynamic> json) {
+    if (json["type"] is! String) {
+      throw Exception(
+        "Unable to decode WsMessageParams. Expected String from \"type\". Received ${json["type"]}}",
+      );
+    }
+    switch (json["type"]) {
+      case "CREATE_ENTITY":
+        return WsMessageParamsCreateEntity.fromJson(json);
+      case "UPDATE_ENTITY":
+        return WsMessageParamsUpdateEntity.fromJson(json);
+      case "DISCONNECT":
+        return WsMessageParamsDisconnect.fromJson(json);
+    }
+    throw Exception(
+      "Unable to decode WsMessageParams. \"${json["type"]}\" doesn't match any of the accepted discriminator values.",
+    );
+  }
+  Map<String, dynamic> toJson();
+}
+
+class WsMessageParamsCreateEntity implements WsMessageParams {
+  @override
+  final String type = "CREATE_ENTITY";
+  final String entityId;
+  final double x;
+  final double y;
+  const WsMessageParamsCreateEntity({
+    required this.entityId,
+    required this.x,
+    required this.y,
+  });
+  factory WsMessageParamsCreateEntity.fromJson(Map<String, dynamic> json) {
+    return WsMessageParamsCreateEntity(
+      entityId: typeFromDynamic<String>(json["entityId"], ""),
+      x: doubleFromDynamic(json["x"], 0),
+      y: doubleFromDynamic(json["y"], 0),
+    );
+  }
+  @override
+  Map<String, dynamic> toJson() {
+    final __result = <String, dynamic>{
+      "type": type,
+      "entityId": entityId,
+      "x": x,
+      "y": y,
+    };
+
+    return __result;
+  }
+
+  WsMessageParamsCreateEntity copyWith({
+    String? entityId,
+    double? x,
+    double? y,
+  }) {
+    return WsMessageParamsCreateEntity(
+      entityId: entityId ?? this.entityId,
+      x: x ?? this.x,
+      y: y ?? this.y,
+    );
+  }
+}
+
+class WsMessageParamsUpdateEntity implements WsMessageParams {
+  @override
+  final String type = "UPDATE_ENTITY";
+  final String entityId;
+  final double x;
+  final double y;
+  const WsMessageParamsUpdateEntity({
+    required this.entityId,
+    required this.x,
+    required this.y,
+  });
+  factory WsMessageParamsUpdateEntity.fromJson(Map<String, dynamic> json) {
+    return WsMessageParamsUpdateEntity(
+      entityId: typeFromDynamic<String>(json["entityId"], ""),
+      x: doubleFromDynamic(json["x"], 0),
+      y: doubleFromDynamic(json["y"], 0),
+    );
+  }
+  @override
+  Map<String, dynamic> toJson() {
+    final __result = <String, dynamic>{
+      "type": type,
+      "entityId": entityId,
+      "x": x,
+      "y": y,
+    };
+
+    return __result;
+  }
+
+  WsMessageParamsUpdateEntity copyWith({
+    String? entityId,
+    double? x,
+    double? y,
+  }) {
+    return WsMessageParamsUpdateEntity(
+      entityId: entityId ?? this.entityId,
+      x: x ?? this.x,
+      y: y ?? this.y,
+    );
+  }
+}
+
+class WsMessageParamsDisconnect implements WsMessageParams {
+  @override
+  final String type = "DISCONNECT";
+  final String reason;
+  const WsMessageParamsDisconnect({
+    required this.reason,
+  });
+  factory WsMessageParamsDisconnect.fromJson(Map<String, dynamic> json) {
+    return WsMessageParamsDisconnect(
+      reason: typeFromDynamic<String>(json["reason"], ""),
+    );
+  }
+  @override
+  Map<String, dynamic> toJson() {
+    final __result = <String, dynamic>{
+      "type": type,
+      "reason": reason,
+    };
+
+    return __result;
+  }
+
+  WsMessageParamsDisconnect copyWith({
+    String? reason,
+  }) {
+    return WsMessageParamsDisconnect(
+      reason: reason ?? this.reason,
+    );
+  }
+}
+
+sealed class WsMessageResponse {
+  final String type;
+  const WsMessageResponse({
+    required this.type,
+  });
+  factory WsMessageResponse.fromJson(Map<String, dynamic> json) {
+    if (json["type"] is! String) {
+      throw Exception(
+        "Unable to decode WsMessageResponse. Expected String from \"type\". Received ${json["type"]}}",
+      );
+    }
+    switch (json["type"]) {
+      case "ENTITY_CREATED":
+        return WsMessageResponseEntityCreated.fromJson(json);
+      case "ENTITY_UPDATED":
+        return WsMessageResponseEntityUpdated.fromJson(json);
+    }
+    throw Exception(
+      "Unable to decode WsMessageResponse. \"${json["type"]}\" doesn't match any of the accepted discriminator values.",
+    );
+  }
+  Map<String, dynamic> toJson();
+}
+
+class WsMessageResponseEntityCreated implements WsMessageResponse {
+  @override
+  final String type = "ENTITY_CREATED";
+  final String entityId;
+  final double x;
+  final double y;
+  const WsMessageResponseEntityCreated({
+    required this.entityId,
+    required this.x,
+    required this.y,
+  });
+  factory WsMessageResponseEntityCreated.fromJson(Map<String, dynamic> json) {
+    return WsMessageResponseEntityCreated(
+      entityId: typeFromDynamic<String>(json["entityId"], ""),
+      x: doubleFromDynamic(json["x"], 0),
+      y: doubleFromDynamic(json["y"], 0),
+    );
+  }
+  @override
+  Map<String, dynamic> toJson() {
+    final __result = <String, dynamic>{
+      "type": type,
+      "entityId": entityId,
+      "x": x,
+      "y": y,
+    };
+
+    return __result;
+  }
+
+  WsMessageResponseEntityCreated copyWith({
+    String? entityId,
+    double? x,
+    double? y,
+  }) {
+    return WsMessageResponseEntityCreated(
+      entityId: entityId ?? this.entityId,
+      x: x ?? this.x,
+      y: y ?? this.y,
+    );
+  }
+}
+
+class WsMessageResponseEntityUpdated implements WsMessageResponse {
+  @override
+  final String type = "ENTITY_UPDATED";
+  final String entityId;
+  final double x;
+  final double y;
+  const WsMessageResponseEntityUpdated({
+    required this.entityId,
+    required this.x,
+    required this.y,
+  });
+  factory WsMessageResponseEntityUpdated.fromJson(Map<String, dynamic> json) {
+    return WsMessageResponseEntityUpdated(
+      entityId: typeFromDynamic<String>(json["entityId"], ""),
+      x: doubleFromDynamic(json["x"], 0),
+      y: doubleFromDynamic(json["y"], 0),
+    );
+  }
+  @override
+  Map<String, dynamic> toJson() {
+    final __result = <String, dynamic>{
+      "type": type,
+      "entityId": entityId,
+      "x": x,
+      "y": y,
+    };
+
+    return __result;
+  }
+
+  WsMessageResponseEntityUpdated copyWith({
+    String? entityId,
+    double? x,
+    double? y,
+  }) {
+    return WsMessageResponseEntityUpdated(
+      entityId: entityId ?? this.entityId,
+      x: x ?? this.x,
+      y: y ?? this.y,
+    );
+  }
+}
diff --git a/tests/clients/client-dart/test/test_client_test.dart b/tests/clients/client-dart/test/test_client_test.dart
index dcc1109c..08486405 100644
--- a/tests/clients/client-dart/test/test_client_test.dart
+++ b/tests/clients/client-dart/test/test_client_test.dart
@@ -407,4 +407,66 @@ Future<void> main() async {
       equals(true),
     );
   });
+  test("[WS] can send/receive objects", () async {
+    final connection = await client.tests.websocketRpc();
+    final messages = <WsMessageResponse>[];
+    final errors = <ArriError>[];
+    final connectionErrors = <ArriError>[];
+    connection.listen(
+      onData: (data) {
+        messages.add(data);
+      },
+      onError: (error) {
+        errors.add(error);
+      },
+      onConnectionError: (error) {
+        print(error);
+      },
+    );
+    connection.send(
+      WsMessageParamsCreateEntity(
+        entityId: "1",
+        x: 10,
+        y: 10,
+      ),
+    );
+    connection.send(
+      WsMessageParamsUpdateEntity(
+        entityId: "2",
+        x: 10,
+        y: 100,
+      ),
+    );
+    connection.send(
+      WsMessageParamsUpdateEntity(
+        entityId: "3",
+        x: 250,
+        y: 50,
+      ),
+    );
+    await Future.delayed(Duration(milliseconds: 500));
+    connection.close();
+    expect(messages.length, equals(3));
+    expect(errors.length, equals(0));
+    expect(connectionErrors.length, equals(0));
+    for (final message in messages) {
+      switch (message) {
+        case WsMessageResponseEntityCreated():
+          expect(message.entityId, equals("1"));
+          expect(message.x, equals(10));
+          expect(message.y, equals(10));
+          break;
+        case WsMessageResponseEntityUpdated():
+          if (message.entityId == "2") {
+            expect(message.x, equals(10));
+            expect(message.y, equals(100));
+          }
+          if (message.entityId == "3") {
+            expect(message.x, equals(250));
+            expect(message.y, equals(50));
+          }
+          break;
+      }
+    }
+  });
 }
diff --git a/tests/clients/client-typescript/testClient.rpc.ts b/tests/clients/client-typescript/testClient.rpc.ts
index 50437f7a..e7109962 100644
--- a/tests/clients/client-typescript/testClient.rpc.ts
+++ b/tests/clients/client-typescript/testClient.rpc.ts
@@ -1,7 +1,13 @@
 /* eslint-disable */
 // @ts-nocheck
 // this file was autogenerated by arri-codegen-ts
-import { arriRequest, arriSseRequest, type SseOptions } from "arri-client";
+import {
+    arriRequest,
+    arriSseRequest,
+    type SseOptions,
+    arriWsRequest,
+    type WsOptions,
+} from "arri-client";
 
 interface TestClientOptions {
     baseUrl?: string;
@@ -239,6 +245,19 @@ export class TestClientTestsService {
             options,
         );
     }
+    websocketRpc(options: WsOptions<WsMessageResponse> = {}) {
+        return arriWsRequest<WsMessageParams, WsMessageResponse>({
+            url: `${this.baseUrl}/rpcs/tests/websocket-rpc`,
+            headers: this.headers,
+            parser: $$WsMessageResponse.parse,
+            serializer: $$WsMessageParams.serialize,
+            onOpen: options.onOpen,
+            onClose: options.onClose,
+            onError: options.onError,
+            onConnectionError: options.onConnectionError,
+            onMessage: options.onMessage,
+        });
+    }
 }
 
 export class TestClientAdaptersService {
@@ -10039,3 +10058,794 @@ export interface ChatMessageUrl {
     date: Date;
     url: string;
 }
+
+export type WsMessageParams =
+    | WsMessageParamsCreateEntity
+    | WsMessageParamsUpdateEntity
+    | WsMessageParamsDisconnect;
+const $$WsMessageParams = {
+    parse(input: Record<any, any>): WsMessageParams {
+        function $fallback(instancePath, schemaPath) {
+            throw new Error(
+                `Error parsing input. InstancePath: "${instancePath}". SchemaPath: "${schemaPath}"`,
+            );
+        }
+
+        if (typeof input === "string") {
+            const json = JSON.parse(input);
+            let result = {};
+            if (typeof json === "object" && json !== null) {
+                switch (json.type) {
+                    case "CREATE_ENTITY": {
+                        if (typeof json === "object" && json !== null) {
+                            const __D1 = {};
+                            __D1.type = "CREATE_ENTITY";
+                            if (typeof json.entityId === "string") {
+                                __D1.entityId = json.entityId;
+                            } else {
+                                $fallback(
+                                    "/entityId",
+                                    "/mapping/properties/entityId/type",
+                                    "Expected string at /entityId",
+                                );
+                            }
+                            if (
+                                typeof json.x === "number" &&
+                                !Number.isNaN(json.x)
+                            ) {
+                                __D1.x = json.x;
+                            } else {
+                                $fallback(
+                                    "/x",
+                                    "/mapping/properties/x/type",
+                                    "Expected number at /x",
+                                );
+                            }
+                            if (
+                                typeof json.y === "number" &&
+                                !Number.isNaN(json.y)
+                            ) {
+                                __D1.y = json.y;
+                            } else {
+                                $fallback(
+                                    "/y",
+                                    "/mapping/properties/y/type",
+                                    "Expected number at /y",
+                                );
+                            }
+                            result = __D1;
+                        } else {
+                            $fallback("", "/mapping", "Expected object");
+                        }
+                        break;
+                    }
+                    case "UPDATE_ENTITY": {
+                        if (typeof json === "object" && json !== null) {
+                            const __D1 = {};
+                            __D1.type = "UPDATE_ENTITY";
+                            if (typeof json.entityId === "string") {
+                                __D1.entityId = json.entityId;
+                            } else {
+                                $fallback(
+                                    "/entityId",
+                                    "/mapping/properties/entityId/type",
+                                    "Expected string at /entityId",
+                                );
+                            }
+                            if (
+                                typeof json.x === "number" &&
+                                !Number.isNaN(json.x)
+                            ) {
+                                __D1.x = json.x;
+                            } else {
+                                $fallback(
+                                    "/x",
+                                    "/mapping/properties/x/type",
+                                    "Expected number at /x",
+                                );
+                            }
+                            if (
+                                typeof json.y === "number" &&
+                                !Number.isNaN(json.y)
+                            ) {
+                                __D1.y = json.y;
+                            } else {
+                                $fallback(
+                                    "/y",
+                                    "/mapping/properties/y/type",
+                                    "Expected number at /y",
+                                );
+                            }
+                            result = __D1;
+                        } else {
+                            $fallback("", "/mapping", "Expected object");
+                        }
+                        break;
+                    }
+                    case "DISCONNECT": {
+                        if (typeof json === "object" && json !== null) {
+                            const __D1 = {};
+                            __D1.type = "DISCONNECT";
+                            if (typeof json.reason === "string") {
+                                __D1.reason = json.reason;
+                            } else {
+                                $fallback(
+                                    "/reason",
+                                    "/mapping/properties/reason/type",
+                                    "Expected string at /reason",
+                                );
+                            }
+                            result = __D1;
+                        } else {
+                            $fallback("", "/mapping", "Expected object");
+                        }
+                        break;
+                    }
+                    default:
+                        $fallback(
+                            "",
+                            "/mapping",
+                            "json.type did not match one of the specified values",
+                        );
+                        break;
+                }
+            } else {
+                $fallback("", "", "Expected Object.");
+            }
+            return result;
+        }
+        let result = {};
+        if (typeof input === "object" && input !== null) {
+            switch (input.type) {
+                case "CREATE_ENTITY": {
+                    if (typeof input === "object" && input !== null) {
+                        const __D1 = {};
+                        __D1.type = "CREATE_ENTITY";
+                        if (typeof input.entityId === "string") {
+                            __D1.entityId = input.entityId;
+                        } else {
+                            $fallback(
+                                "/entityId",
+                                "/mapping/properties/entityId/type",
+                                "Expected string at /entityId",
+                            );
+                        }
+                        if (
+                            typeof input.x === "number" &&
+                            !Number.isNaN(input.x)
+                        ) {
+                            __D1.x = input.x;
+                        } else {
+                            $fallback(
+                                "/x",
+                                "/mapping/properties/x/type",
+                                "Expected number at /x",
+                            );
+                        }
+                        if (
+                            typeof input.y === "number" &&
+                            !Number.isNaN(input.y)
+                        ) {
+                            __D1.y = input.y;
+                        } else {
+                            $fallback(
+                                "/y",
+                                "/mapping/properties/y/type",
+                                "Expected number at /y",
+                            );
+                        }
+                        result = __D1;
+                    } else {
+                        $fallback("", "/mapping", "Expected object");
+                    }
+                    break;
+                }
+                case "UPDATE_ENTITY": {
+                    if (typeof input === "object" && input !== null) {
+                        const __D1 = {};
+                        __D1.type = "UPDATE_ENTITY";
+                        if (typeof input.entityId === "string") {
+                            __D1.entityId = input.entityId;
+                        } else {
+                            $fallback(
+                                "/entityId",
+                                "/mapping/properties/entityId/type",
+                                "Expected string at /entityId",
+                            );
+                        }
+                        if (
+                            typeof input.x === "number" &&
+                            !Number.isNaN(input.x)
+                        ) {
+                            __D1.x = input.x;
+                        } else {
+                            $fallback(
+                                "/x",
+                                "/mapping/properties/x/type",
+                                "Expected number at /x",
+                            );
+                        }
+                        if (
+                            typeof input.y === "number" &&
+                            !Number.isNaN(input.y)
+                        ) {
+                            __D1.y = input.y;
+                        } else {
+                            $fallback(
+                                "/y",
+                                "/mapping/properties/y/type",
+                                "Expected number at /y",
+                            );
+                        }
+                        result = __D1;
+                    } else {
+                        $fallback("", "/mapping", "Expected object");
+                    }
+                    break;
+                }
+                case "DISCONNECT": {
+                    if (typeof input === "object" && input !== null) {
+                        const __D1 = {};
+                        __D1.type = "DISCONNECT";
+                        if (typeof input.reason === "string") {
+                            __D1.reason = input.reason;
+                        } else {
+                            $fallback(
+                                "/reason",
+                                "/mapping/properties/reason/type",
+                                "Expected string at /reason",
+                            );
+                        }
+                        result = __D1;
+                    } else {
+                        $fallback("", "/mapping", "Expected object");
+                    }
+                    break;
+                }
+                default:
+                    $fallback(
+                        "",
+                        "/mapping",
+                        "input.type did not match one of the specified values",
+                    );
+                    break;
+            }
+        } else {
+            $fallback("", "", "Expected Object.");
+        }
+        return result;
+    },
+    serialize(input: WsMessageParams): string {
+        let json = "";
+
+        const STR_ESCAPE =
+            /[\u0000-\u001f\u0022\u005c\ud800-\udfff]|[\ud800-\udbff](?![\udc00-\udfff])|(?:[^\ud800-\udbff]|^)[\udc00-\udfff]/;
+        switch (input.type) {
+            case "CREATE_ENTITY": {
+                json += "";
+                json += "{";
+                json += `"type":"CREATE_ENTITY"`;
+                json += `,"entityId":`;
+                if (input.entityId.length < 42) {
+                    let __result__ = "";
+                    let __last__ = -1;
+                    let __point__ = 255;
+                    let __finished__ = false;
+                    for (let i = 0; i < input.entityId.length; i++) {
+                        __point__ = input.entityId.charCodeAt(i);
+                        if (
+                            __point__ < 32 ||
+                            (__point__ >= 0xd800 && __point__ <= 0xdfff)
+                        ) {
+                            json += JSON.stringify(input.entityId);
+                            __finished__ = true;
+                            break;
+                        }
+                        if (__point__ === 0x22 || __point__ === 0x5c) {
+                            __last__ === -1 && (__last__ = 0);
+                            __result__ +=
+                                input.entityId.slice(__last__, i) + "\\";
+                            __last__ = i;
+                        }
+                    }
+                    if (!__finished__) {
+                        if (__last__ === -1) {
+                            json += `"${input.entityId}"`;
+                        } else {
+                            json += `"${__result__}${input.entityId.slice(__last__)}"`;
+                        }
+                    }
+                } else if (
+                    input.entityId.length < 5000 &&
+                    !STR_ESCAPE.test(input.entityId)
+                ) {
+                    json += `"${input.entityId}"`;
+                } else {
+                    json += JSON.stringify(input.entityId);
+                }
+
+                if (Number.isNaN(input.x)) {
+                    throw new Error("Expected number at /x got NaN");
+                }
+                json += `,"x":${input.x}`;
+
+                if (Number.isNaN(input.y)) {
+                    throw new Error("Expected number at /y got NaN");
+                }
+                json += `,"y":${input.y}`;
+                json += "}";
+                break;
+            }
+            case "UPDATE_ENTITY": {
+                json += "";
+                json += "{";
+                json += `"type":"UPDATE_ENTITY"`;
+                json += `,"entityId":`;
+                if (input.entityId.length < 42) {
+                    let __result__ = "";
+                    let __last__ = -1;
+                    let __point__ = 255;
+                    let __finished__ = false;
+                    for (let i = 0; i < input.entityId.length; i++) {
+                        __point__ = input.entityId.charCodeAt(i);
+                        if (
+                            __point__ < 32 ||
+                            (__point__ >= 0xd800 && __point__ <= 0xdfff)
+                        ) {
+                            json += JSON.stringify(input.entityId);
+                            __finished__ = true;
+                            break;
+                        }
+                        if (__point__ === 0x22 || __point__ === 0x5c) {
+                            __last__ === -1 && (__last__ = 0);
+                            __result__ +=
+                                input.entityId.slice(__last__, i) + "\\";
+                            __last__ = i;
+                        }
+                    }
+                    if (!__finished__) {
+                        if (__last__ === -1) {
+                            json += `"${input.entityId}"`;
+                        } else {
+                            json += `"${__result__}${input.entityId.slice(__last__)}"`;
+                        }
+                    }
+                } else if (
+                    input.entityId.length < 5000 &&
+                    !STR_ESCAPE.test(input.entityId)
+                ) {
+                    json += `"${input.entityId}"`;
+                } else {
+                    json += JSON.stringify(input.entityId);
+                }
+
+                if (Number.isNaN(input.x)) {
+                    throw new Error("Expected number at /x got NaN");
+                }
+                json += `,"x":${input.x}`;
+
+                if (Number.isNaN(input.y)) {
+                    throw new Error("Expected number at /y got NaN");
+                }
+                json += `,"y":${input.y}`;
+                json += "}";
+                break;
+            }
+            case "DISCONNECT": {
+                json += "";
+                json += "{";
+                json += `"type":"DISCONNECT"`;
+                json += `,"reason":`;
+                if (input.reason.length < 42) {
+                    let __result__ = "";
+                    let __last__ = -1;
+                    let __point__ = 255;
+                    let __finished__ = false;
+                    for (let i = 0; i < input.reason.length; i++) {
+                        __point__ = input.reason.charCodeAt(i);
+                        if (
+                            __point__ < 32 ||
+                            (__point__ >= 0xd800 && __point__ <= 0xdfff)
+                        ) {
+                            json += JSON.stringify(input.reason);
+                            __finished__ = true;
+                            break;
+                        }
+                        if (__point__ === 0x22 || __point__ === 0x5c) {
+                            __last__ === -1 && (__last__ = 0);
+                            __result__ +=
+                                input.reason.slice(__last__, i) + "\\";
+                            __last__ = i;
+                        }
+                    }
+                    if (!__finished__) {
+                        if (__last__ === -1) {
+                            json += `"${input.reason}"`;
+                        } else {
+                            json += `"${__result__}${input.reason.slice(__last__)}"`;
+                        }
+                    }
+                } else if (
+                    input.reason.length < 5000 &&
+                    !STR_ESCAPE.test(input.reason)
+                ) {
+                    json += `"${input.reason}"`;
+                } else {
+                    json += JSON.stringify(input.reason);
+                }
+                json += "}";
+                break;
+            }
+        }
+        return json;
+    },
+};
+export interface WsMessageParamsCreateEntity {
+    type: "CREATE_ENTITY";
+    entityId: string;
+    x: number;
+    y: number;
+}
+
+export interface WsMessageParamsUpdateEntity {
+    type: "UPDATE_ENTITY";
+    entityId: string;
+    x: number;
+    y: number;
+}
+
+export interface WsMessageParamsDisconnect {
+    type: "DISCONNECT";
+    reason: string;
+}
+
+export type WsMessageResponse =
+    | WsMessageResponseEntityCreated
+    | WsMessageResponseEntityUpdated;
+const $$WsMessageResponse = {
+    parse(input: Record<any, any>): WsMessageResponse {
+        function $fallback(instancePath, schemaPath) {
+            throw new Error(
+                `Error parsing input. InstancePath: "${instancePath}". SchemaPath: "${schemaPath}"`,
+            );
+        }
+
+        if (typeof input === "string") {
+            const json = JSON.parse(input);
+            let result = {};
+            if (typeof json === "object" && json !== null) {
+                switch (json.type) {
+                    case "ENTITY_CREATED": {
+                        if (typeof json === "object" && json !== null) {
+                            const __D1 = {};
+                            __D1.type = "ENTITY_CREATED";
+                            if (typeof json.entityId === "string") {
+                                __D1.entityId = json.entityId;
+                            } else {
+                                $fallback(
+                                    "/entityId",
+                                    "/mapping/properties/entityId/type",
+                                    "Expected string at /entityId",
+                                );
+                            }
+                            if (
+                                typeof json.x === "number" &&
+                                !Number.isNaN(json.x)
+                            ) {
+                                __D1.x = json.x;
+                            } else {
+                                $fallback(
+                                    "/x",
+                                    "/mapping/properties/x/type",
+                                    "Expected number at /x",
+                                );
+                            }
+                            if (
+                                typeof json.y === "number" &&
+                                !Number.isNaN(json.y)
+                            ) {
+                                __D1.y = json.y;
+                            } else {
+                                $fallback(
+                                    "/y",
+                                    "/mapping/properties/y/type",
+                                    "Expected number at /y",
+                                );
+                            }
+                            result = __D1;
+                        } else {
+                            $fallback("", "/mapping", "Expected object");
+                        }
+                        break;
+                    }
+                    case "ENTITY_UPDATED": {
+                        if (typeof json === "object" && json !== null) {
+                            const __D1 = {};
+                            __D1.type = "ENTITY_UPDATED";
+                            if (typeof json.entityId === "string") {
+                                __D1.entityId = json.entityId;
+                            } else {
+                                $fallback(
+                                    "/entityId",
+                                    "/mapping/properties/entityId/type",
+                                    "Expected string at /entityId",
+                                );
+                            }
+                            if (
+                                typeof json.x === "number" &&
+                                !Number.isNaN(json.x)
+                            ) {
+                                __D1.x = json.x;
+                            } else {
+                                $fallback(
+                                    "/x",
+                                    "/mapping/properties/x/type",
+                                    "Expected number at /x",
+                                );
+                            }
+                            if (
+                                typeof json.y === "number" &&
+                                !Number.isNaN(json.y)
+                            ) {
+                                __D1.y = json.y;
+                            } else {
+                                $fallback(
+                                    "/y",
+                                    "/mapping/properties/y/type",
+                                    "Expected number at /y",
+                                );
+                            }
+                            result = __D1;
+                        } else {
+                            $fallback("", "/mapping", "Expected object");
+                        }
+                        break;
+                    }
+                    default:
+                        $fallback(
+                            "",
+                            "/mapping",
+                            "json.type did not match one of the specified values",
+                        );
+                        break;
+                }
+            } else {
+                $fallback("", "", "Expected Object.");
+            }
+            return result;
+        }
+        let result = {};
+        if (typeof input === "object" && input !== null) {
+            switch (input.type) {
+                case "ENTITY_CREATED": {
+                    if (typeof input === "object" && input !== null) {
+                        const __D1 = {};
+                        __D1.type = "ENTITY_CREATED";
+                        if (typeof input.entityId === "string") {
+                            __D1.entityId = input.entityId;
+                        } else {
+                            $fallback(
+                                "/entityId",
+                                "/mapping/properties/entityId/type",
+                                "Expected string at /entityId",
+                            );
+                        }
+                        if (
+                            typeof input.x === "number" &&
+                            !Number.isNaN(input.x)
+                        ) {
+                            __D1.x = input.x;
+                        } else {
+                            $fallback(
+                                "/x",
+                                "/mapping/properties/x/type",
+                                "Expected number at /x",
+                            );
+                        }
+                        if (
+                            typeof input.y === "number" &&
+                            !Number.isNaN(input.y)
+                        ) {
+                            __D1.y = input.y;
+                        } else {
+                            $fallback(
+                                "/y",
+                                "/mapping/properties/y/type",
+                                "Expected number at /y",
+                            );
+                        }
+                        result = __D1;
+                    } else {
+                        $fallback("", "/mapping", "Expected object");
+                    }
+                    break;
+                }
+                case "ENTITY_UPDATED": {
+                    if (typeof input === "object" && input !== null) {
+                        const __D1 = {};
+                        __D1.type = "ENTITY_UPDATED";
+                        if (typeof input.entityId === "string") {
+                            __D1.entityId = input.entityId;
+                        } else {
+                            $fallback(
+                                "/entityId",
+                                "/mapping/properties/entityId/type",
+                                "Expected string at /entityId",
+                            );
+                        }
+                        if (
+                            typeof input.x === "number" &&
+                            !Number.isNaN(input.x)
+                        ) {
+                            __D1.x = input.x;
+                        } else {
+                            $fallback(
+                                "/x",
+                                "/mapping/properties/x/type",
+                                "Expected number at /x",
+                            );
+                        }
+                        if (
+                            typeof input.y === "number" &&
+                            !Number.isNaN(input.y)
+                        ) {
+                            __D1.y = input.y;
+                        } else {
+                            $fallback(
+                                "/y",
+                                "/mapping/properties/y/type",
+                                "Expected number at /y",
+                            );
+                        }
+                        result = __D1;
+                    } else {
+                        $fallback("", "/mapping", "Expected object");
+                    }
+                    break;
+                }
+                default:
+                    $fallback(
+                        "",
+                        "/mapping",
+                        "input.type did not match one of the specified values",
+                    );
+                    break;
+            }
+        } else {
+            $fallback("", "", "Expected Object.");
+        }
+        return result;
+    },
+    serialize(input: WsMessageResponse): string {
+        let json = "";
+
+        const STR_ESCAPE =
+            /[\u0000-\u001f\u0022\u005c\ud800-\udfff]|[\ud800-\udbff](?![\udc00-\udfff])|(?:[^\ud800-\udbff]|^)[\udc00-\udfff]/;
+        switch (input.type) {
+            case "ENTITY_CREATED": {
+                json += "";
+                json += "{";
+                json += `"type":"ENTITY_CREATED"`;
+                json += `,"entityId":`;
+                if (input.entityId.length < 42) {
+                    let __result__ = "";
+                    let __last__ = -1;
+                    let __point__ = 255;
+                    let __finished__ = false;
+                    for (let i = 0; i < input.entityId.length; i++) {
+                        __point__ = input.entityId.charCodeAt(i);
+                        if (
+                            __point__ < 32 ||
+                            (__point__ >= 0xd800 && __point__ <= 0xdfff)
+                        ) {
+                            json += JSON.stringify(input.entityId);
+                            __finished__ = true;
+                            break;
+                        }
+                        if (__point__ === 0x22 || __point__ === 0x5c) {
+                            __last__ === -1 && (__last__ = 0);
+                            __result__ +=
+                                input.entityId.slice(__last__, i) + "\\";
+                            __last__ = i;
+                        }
+                    }
+                    if (!__finished__) {
+                        if (__last__ === -1) {
+                            json += `"${input.entityId}"`;
+                        } else {
+                            json += `"${__result__}${input.entityId.slice(__last__)}"`;
+                        }
+                    }
+                } else if (
+                    input.entityId.length < 5000 &&
+                    !STR_ESCAPE.test(input.entityId)
+                ) {
+                    json += `"${input.entityId}"`;
+                } else {
+                    json += JSON.stringify(input.entityId);
+                }
+
+                if (Number.isNaN(input.x)) {
+                    throw new Error("Expected number at /x got NaN");
+                }
+                json += `,"x":${input.x}`;
+
+                if (Number.isNaN(input.y)) {
+                    throw new Error("Expected number at /y got NaN");
+                }
+                json += `,"y":${input.y}`;
+                json += "}";
+                break;
+            }
+            case "ENTITY_UPDATED": {
+                json += "";
+                json += "{";
+                json += `"type":"ENTITY_UPDATED"`;
+                json += `,"entityId":`;
+                if (input.entityId.length < 42) {
+                    let __result__ = "";
+                    let __last__ = -1;
+                    let __point__ = 255;
+                    let __finished__ = false;
+                    for (let i = 0; i < input.entityId.length; i++) {
+                        __point__ = input.entityId.charCodeAt(i);
+                        if (
+                            __point__ < 32 ||
+                            (__point__ >= 0xd800 && __point__ <= 0xdfff)
+                        ) {
+                            json += JSON.stringify(input.entityId);
+                            __finished__ = true;
+                            break;
+                        }
+                        if (__point__ === 0x22 || __point__ === 0x5c) {
+                            __last__ === -1 && (__last__ = 0);
+                            __result__ +=
+                                input.entityId.slice(__last__, i) + "\\";
+                            __last__ = i;
+                        }
+                    }
+                    if (!__finished__) {
+                        if (__last__ === -1) {
+                            json += `"${input.entityId}"`;
+                        } else {
+                            json += `"${__result__}${input.entityId.slice(__last__)}"`;
+                        }
+                    }
+                } else if (
+                    input.entityId.length < 5000 &&
+                    !STR_ESCAPE.test(input.entityId)
+                ) {
+                    json += `"${input.entityId}"`;
+                } else {
+                    json += JSON.stringify(input.entityId);
+                }
+
+                if (Number.isNaN(input.x)) {
+                    throw new Error("Expected number at /x got NaN");
+                }
+                json += `,"x":${input.x}`;
+
+                if (Number.isNaN(input.y)) {
+                    throw new Error("Expected number at /y got NaN");
+                }
+                json += `,"y":${input.y}`;
+                json += "}";
+                break;
+            }
+        }
+        return json;
+    },
+};
+export interface WsMessageResponseEntityCreated {
+    type: "ENTITY_CREATED";
+    entityId: string;
+    x: number;
+    y: number;
+}
+
+export interface WsMessageResponseEntityUpdated {
+    type: "ENTITY_UPDATED";
+    entityId: string;
+    x: number;
+    y: number;
+}
diff --git a/tests/clients/client-typescript/testClient.test.ts b/tests/clients/client-typescript/testClient.test.ts
index 0af382c8..c231f239 100644
--- a/tests/clients/client-typescript/testClient.test.ts
+++ b/tests/clients/client-typescript/testClient.test.ts
@@ -6,6 +6,7 @@ import {
     type ObjectWithEveryType,
     type ObjectWithEveryOptionalType,
     type ObjectWithEveryNullableType,
+    type WsMessageResponse,
     type RecursiveObject,
     type RecursiveUnion,
     type TypeBoxObject,
@@ -365,6 +366,65 @@ test("[SSE] auto-reconnects when connection is closed by server", async () => {
     controller.abort();
 });
 
+test("Websocket Requests", async () => {
+    let connectionCount = 0;
+    let messageCount = 0;
+    const errorCount = 0;
+    const msgMap: Record<string, WsMessageResponse> = {};
+    const controller = client.tests.websocketRpc({
+        onMessage(msg) {
+            messageCount++;
+            msgMap[msg.entityId] = msg;
+        },
+        onConnectionError(err) {
+            console.error(err);
+            throw new ArriErrorInstance({
+                code: err.code,
+                message: err.message,
+                data: err.data,
+                stack: err.stack,
+            });
+        },
+    });
+    controller.onOpen = () => {
+        connectionCount++;
+        controller.send({
+            type: "CREATE_ENTITY",
+            entityId: "1",
+            x: 100,
+            y: 200,
+        });
+        controller.send({
+            type: "UPDATE_ENTITY",
+            entityId: "2",
+            x: 1,
+            y: 2,
+        });
+        controller.send({
+            type: "UPDATE_ENTITY",
+            entityId: "3",
+            x: 5,
+            y: -5,
+        });
+    };
+    controller.connect();
+    await new Promise((resolve) => {
+        setTimeout(() => {
+            resolve(true);
+        }, 2000);
+    });
+    controller.close();
+    expect(connectionCount).toBe(1);
+    expect(messageCount).toBe(3);
+    expect(errorCount).toBe(0);
+    expect(msgMap["1"]!.x).toBe(100);
+    expect(msgMap["1"]!.y).toBe(200);
+    expect(msgMap["2"]!.x).toBe(1);
+    expect(msgMap["2"]!.y).toBe(2);
+    expect(msgMap["3"]!.x).toBe(5);
+    expect(msgMap["3"]!.y).toBe(-5);
+});
+
 describe("arri adapters", () => {
     test("typebox adapter", async () => {
         const input: TypeBoxObject = {
diff --git a/tests/server/src/procedures/tests/websocketRpc.rpc.ts b/tests/server/src/procedures/tests/websocketRpc.rpc.ts
new file mode 100644
index 00000000..89b15951
--- /dev/null
+++ b/tests/server/src/procedures/tests/websocketRpc.rpc.ts
@@ -0,0 +1,68 @@
+import { defineWebsocketRpc } from "arri";
+import { a } from "arri-validate";
+
+const EntityFields = a.object({
+    entityId: a.string(),
+    x: a.float64(),
+    y: a.float64(),
+});
+
+const ParamsSchema = a.discriminator(
+    "type",
+    {
+        CREATE_ENTITY: EntityFields,
+        UPDATE_ENTITY: EntityFields,
+        DISCONNECT: a.object({
+            reason: a.string(),
+        }),
+    },
+    {
+        id: "WsMessageParams",
+    },
+);
+
+const ResponseSchema = a.discriminator(
+    "type",
+    {
+        ENTITY_CREATED: EntityFields,
+        ENTITY_UPDATED: EntityFields,
+    },
+    {
+        id: "WsMessageResponse",
+    },
+);
+
+export default defineWebsocketRpc({
+    params: ParamsSchema,
+    response: ResponseSchema,
+    handler: {
+        onOpen: (peer) => {},
+        onClose: (peer) => {},
+        onMessage: async (peer, message) => {
+            if (!peer.context.messageCount) {
+                peer.context.messageCount = 1;
+            } else {
+                peer.context.messageCount += 1;
+            }
+            switch (message.type) {
+                case "CREATE_ENTITY":
+                    peer.send({
+                        type: "ENTITY_CREATED",
+                        entityId: message.entityId,
+                        x: message.x,
+                        y: message.y,
+                    });
+                    break;
+                case "UPDATE_ENTITY":
+                    peer.send({
+                        type: "ENTITY_UPDATED",
+                        entityId: message.entityId,
+                        x: message.x,
+                        y: message.y,
+                    });
+                    break;
+            }
+        },
+        onError: (peer, error) => {},
+    },
+});
diff --git a/tests/server/src/server.ts b/tests/server/src/server.ts
index 777b6571..c9546034 100644
--- a/tests/server/src/server.ts
+++ b/tests/server/src/server.ts
@@ -4,6 +4,18 @@ import app from "virtual:arri/app";
 
 void listen(toNodeListener(app.h3App), {
     port: process.env.PORT ?? 2020,
+    ws: {
+        resolve(info) {
+            if (app.h3App.websocket.resolve) {
+                return app.h3App.websocket.resolve(info);
+            }
+            return (
+                app.h3App.websocket.hooks ??
+                app.h3App.handler.__websocket__ ??
+                {}
+            );
+        },
+    },
     public: true,
     http2: true,
 });