Python RPC Client (experimental)

Python client can be generated with vovk generate command using py template or pySrc template.

Generate Python package with the CLI:


npx vovk generate --from py --out ./python_package


pnpm dlx vovk generate --from py --out ./python_package


yarn dlx vovk generate --from py --out ./python_package


bun x vovk generate --from py --out ./python_package

This will generate a Python package with the following structure:

- - __init__.py
  - api_client.py
  - py.typed
  - schema.json
- pyproject.toml
- setup.cfg
- README.md

The package is ready to be published to PyPI with the following command.


python3 -m build ./python_package --wheel --sdist && python3 -m twine upload ./python_package/dist/*

In case if you want to generate Python source only, to be used as part of another Python project, you can use pySrc template:


npx vovk generate --from pySrc --out ./python_src


pnpm dlx vovk generate --from pySrc --out ./python_src


yarn dlx vovk generate --from pySrc --out ./python_src


bun x vovk generate --from pySrc --out ./python_src

This will generate a Python source with the following structure:

- __init__.py
- api_client.py
- py.typed
- schema.json

Configuring the Python Client

The generation can be configured, so the client will be generated automatically with generate command with no flags but also when vovk dev is run, that performs “hot generation” whenever the schema is changed. In order to do that, you need to add py template to the composed client config:

vovk.config.mjs


/** @type {import('vovk').VovkConfig} */
const config = {
  composedClient: {
    fromTemplates: ['cjs', 'mjs', 'py'], // keep the default "cjs" and "mjs" templates
  },
};
export default config;

The py template as well as some other templates has default outDir (by default equals to ./dist_python) for the composed client configuration, that can be changed by defining a template definition in the config:

vovk.config.mjs


/** @type {import('vovk').VovkConfig} */
const config = {
  // ...
  clientTemplateDefs: {
    py: {
      extends: 'py', // extends the built-in "py" template
      composedClient: {
        outDir: './my_dist_python', // updates the out directory for the composed client
      },
    },
  },
};
export default config;

Generated Python Client Example

JSON endpoints

All the code below was copied with slight modifications from the real example, explained at Hello World page.

A controller like this:

UserController.ts

UserService.ts

1import { z } from "zod";
2import { prefix, post, operation } from "vovk";
3import { withZod } from "vovk-zod";
4import UserService from "./UserService";
5
6@prefix("users")
7export default class UserController {
8  @operation({
9    summary: "Update user",
10    description: "Update user by ID",
11  })
12  @post("{id}")
13  static updateUser = withZod({
14    body: z
15      .object({
16        email: z.email().meta({
17          description: "User email",
18          examples: ["john@example.com", "jane@example.com"],
19        }),
20        profile: z
21          .object({
22            name: z.string().meta({
23              description: "User full name",
24              examples: ["John Doe", "Jane Smith"],
25            }),
26            age: z
27              .int()
28              .min(16)
29              .max(120)
30              .meta({ description: "User age", examples: [25, 30] }),
31          })
32          .meta({ description: "User profile object" }),
33      })
34      .meta({ description: "User data object" }),
35    params: z
36      .object({
37        id: z.uuid().meta({
38          description: "User ID",
39          examples: ["123e4567-e89b-12d3-a456-426614174000"],
40        }),
41      })
42      .meta({
43        description: "Path parameters",
44      }),
45    query: z
46      .object({
47        notify: z
48          .enum(["email", "push", "none"])
49          .meta({ description: "Notification type" }),
50      })
51      .meta({
52        description: "Query parameters",
53      }),
54    output: z
55      .object({
56        success: z.boolean().meta({ description: "Success status" }),
57      })
58      .meta({ description: "Response object" }),
59    async handle(req, { id }) {
60      const body = await req.json();
61      const notify = req.nextUrl.searchParams.get("notify");
62
63      return UserService.updateUser(id, body, notify);
64    },
65  });
66}

1import { z } from "zod";
2import { prefix, post, operation } from "vovk";
3import { withZod } from "vovk-zod";
4import UserService from "./UserService";
5
6@prefix("users")
7export default class UserController {
8  @operation({
9    summary: "Update user",
10    description: "Update user by ID",
11  })
12  @post("{id}")
13  static updateUser = withZod({
14    body: z
15      .object({
16        email: z.email().meta({
17          description: "User email",
18          examples: ["john@example.com", "jane@example.com"],
19        }),
20        profile: z
21          .object({
22            name: z.string().meta({
23              description: "User full name",
24              examples: ["John Doe", "Jane Smith"],
25            }),
26            age: z
27              .int()
28              .min(16)
29              .max(120)
30              .meta({ description: "User age", examples: [25, 30] }),
31          })
32          .meta({ description: "User profile object" }),
33      })
34      .meta({ description: "User data object" }),
35    params: z
36      .object({
37        id: z.uuid().meta({
38          description: "User ID",
39          examples: ["123e4567-e89b-12d3-a456-426614174000"],
40        }),
41      })
42      .meta({
43        description: "Path parameters",
44      }),
45    query: z
46      .object({
47        notify: z
48          .enum(["email", "push", "none"])
49          .meta({ description: "Notification type" }),
50      })
51      .meta({
52        description: "Query parameters",
53      }),
54    output: z
55      .object({
56        success: z.boolean().meta({ description: "Success status" }),
57      })
58      .meta({ description: "Response object" }),
59    async handle(req, { id }) {
60      const body = await req.json();
61      const notify = req.nextUrl.searchParams.get("notify");
62
63      return UserService.updateUser(id, body, notify);
64    },
65  });
66}

1import { z } from "zod";
2import { prefix, post, operation } from "vovk";
3import { withZod } from "vovk-zod";
4import UserService from "./UserService";
5
6@prefix("users")
7export default class UserController {
8  @operation({
9    summary: "Update user",
10    description: "Update user by ID",
11  })
12  @post("{id}")
13  static updateUser = withZod({
14    body: z
15      .object({
16        email: z.email().meta({
17          description: "User email",
18          examples: ["john@example.com", "jane@example.com"],
19        }),
20        profile: z
21          .object({
22            name: z.string().meta({
23              description: "User full name",
24              examples: ["John Doe", "Jane Smith"],
25            }),
26            age: z
27              .int()
28              .min(16)
29              .max(120)
30              .meta({ description: "User age", examples: [25, 30] }),
31          })
32          .meta({ description: "User profile object" }),
33      })
34      .meta({ description: "User data object" }),
35    params: z
36      .object({
37        id: z.uuid().meta({
38          description: "User ID",
39          examples: ["123e4567-e89b-12d3-a456-426614174000"],
40        }),
41      })
42      .meta({
43        description: "Path parameters",
44      }),
45    query: z
46      .object({
47        notify: z
48          .enum(["email", "push", "none"])
49          .meta({ description: "Notification type" }),
50      })
51      .meta({
52        description: "Query parameters",
53      }),
54    output: z
55      .object({
56        success: z.boolean().meta({ description: "Success status" }),
57      })
58      .meta({ description: "Response object" }),
59    async handle(req, { id }) {
60      const body = await req.json();
61      const notify = req.nextUrl.searchParams.get("notify");
62
63      return UserService.updateUser(id, body, notify);
64    },
65  });
66}

1import { z } from "zod";
2import { prefix, post, operation } from "vovk";
3import { withZod } from "vovk-zod";
4import UserService from "./UserService";
5
6@prefix("users")
7export default class UserController {
8  @operation({
9    summary: "Update user",
10    description: "Update user by ID",
11  })
12  @post("{id}")
13  static updateUser = withZod({
14    body: z
15      .object({
16        email: z.email().meta({
17          description: "User email",
18          examples: ["john@example.com", "jane@example.com"],
19        }),
20        profile: z
21          .object({
22            name: z.string().meta({
23              description: "User full name",
24              examples: ["John Doe", "Jane Smith"],
25            }),
26            age: z
27              .int()
28              .min(16)
29              .max(120)
30              .meta({ description: "User age", examples: [25, 30] }),
31          })
32          .meta({ description: "User profile object" }),
33      })
34      .meta({ description: "User data object" }),
35    params: z
36      .object({
37        id: z.uuid().meta({
38          description: "User ID",
39          examples: ["123e4567-e89b-12d3-a456-426614174000"],
40        }),
41      })
42      .meta({
43        description: "Path parameters",
44      }),
45    query: z
46      .object({
47        notify: z
48          .enum(["email", "push", "none"])
49          .meta({ description: "Notification type" }),
50      })
51      .meta({
52        description: "Query parameters",
53      }),
54    output: z
55      .object({
56        success: z.boolean().meta({ description: "Success status" }),
57      })
58      .meta({ description: "Response object" }),
59    async handle(req, { id }) {
60      const body = await req.json();
61      const notify = req.nextUrl.searchParams.get("notify");
62
63      return UserService.updateUser(id, body, notify);
64    },
65  });
66}

1import type { VovkBody, VovkOutput, VovkParams, VovkQuery } from "vovk";
2import type UserController from "./UserController";
3
4export default class UserService {
5  static updateUser = (
6    id: VovkParams<typeof UserController.updateUser>["id"],
7    body: VovkBody<typeof UserController.updateUser>,
8    notify: VovkQuery<typeof UserController.updateUser>["notify"],
9  ) => {
10    console.log(
11      id satisfies string,
12      body satisfies { email: string; profile: { name: string; age: number } },
13      notify satisfies "email" | "push" | "none",
14    );
15    return {
16      success: true,
17    } satisfies VovkOutput<typeof UserController.updateUser>;
18  };
19}

1import type { VovkBody, VovkOutput, VovkParams, VovkQuery } from "vovk";
2import type UserController from "./UserController";
3
4export default class UserService {
5  static updateUser = (
6    id: VovkParams<typeof UserController.updateUser>["id"],
7    body: VovkBody<typeof UserController.updateUser>,
8    notify: VovkQuery<typeof UserController.updateUser>["notify"],
9  ) => {
10    console.log(
11      id satisfies string,
12      body satisfies { email: string; profile: { name: string; age: number } },
13      notify satisfies "email" | "push" | "none",
14    );
15    return {
16      success: true,
17    } satisfies VovkOutput<typeof UserController.updateUser>;
18  };
19}

1import type { VovkBody, VovkOutput, VovkParams, VovkQuery } from "vovk";
2import type UserController from "./UserController";
3
4export default class UserService {
5  static updateUser = (
6    id: VovkParams<typeof UserController.updateUser>["id"],
7    body: VovkBody<typeof UserController.updateUser>,
8    notify: VovkQuery<typeof UserController.updateUser>["notify"],
9  ) => {
10    console.log(
11      id satisfies string,
12      body satisfies { email: string; profile: { name: string; age: number } },
13      notify satisfies "email" | "push" | "none",
14    );
15    return {
16      success: true,
17    } satisfies VovkOutput<typeof UserController.updateUser>;
18  };
19}

1import type { VovkBody, VovkOutput, VovkParams, VovkQuery } from "vovk";
2import type UserController from "./UserController";
3
4export default class UserService {
5  static updateUser = (
6    id: VovkParams<typeof UserController.updateUser>["id"],
7    body: VovkBody<typeof UserController.updateUser>,
8    notify: VovkQuery<typeof UserController.updateUser>["notify"],
9  ) => {
10    console.log(
11      id satisfies string,
12      body satisfies { email: string; profile: { name: string; age: number } },
13      notify satisfies "email" | "push" | "none",
14    );
15    return {
16      success: true,
17    } satisfies VovkOutput<typeof UserController.updateUser>;
18  };
19}

… will emit Vovk.ts Schema, that in its turn will be used to generate the Python client, following the language conventions where possible, adding comments defined at schema description, and generating the most suitable types. For example, the age is generated as int.

./dist_python/src/package_name/__init__.py


from __future__ import annotations
from typing import Any, Dict, List, Literal, Optional, Set, TypedDict, Union, Tuple, Generator # type: ignore
from .api_client import ApiClient, HttpException
 
HttpException = HttpException
 
client = ApiClient('https://vovk-hello-world.vercel.app/api')
 
class UserRPC: 
    # UserRPC.update_user POST `https://vovk-hello-world.vercel.app/api/users/{id}`
    class __UpdateUserBody_profile(TypedDict):
        """
        User profile object
        """
        name: str
        age: int
    class UpdateUserBody(TypedDict):
        """
        User data object
        """
        email: str
        profile: UserRPC.__UpdateUserBody_profile
    class UpdateUserQuery(TypedDict):
        """
        Query parameters
        """
        notify: Literal["email", "push", "none"]
    class UpdateUserParams(TypedDict):
        """
        Path parameters
        """
        id: str
    class UpdateUserOutput(TypedDict):
        """
        Response object
        """
        success: bool
    @staticmethod
    def update_user(
        body: UpdateUserBody,
        query: UpdateUserQuery,
        params: UpdateUserParams,
        headers: Optional[Dict[str, str]] = None,
        files: Optional[Dict[str, Any]] = None,
        api_root: Optional[str] = None,
        disable_client_validation: bool = False
    ) -> UpdateUserOutput:
        """ 
        Update user
        Description: Update user by ID
        Body: User data object
        Query: Query parameters
        Returns: Response object
        """
        return client.request( # type: ignore
            segment_name='',
            rpc_name='UserRPC',
            handler_name='updateUser',
            body=body,
            query=query,
            params=params,
            headers=headers,
            files=files,
            api_root=api_root,
            disable_client_validation=disable_client_validation
        )

All RPC modules are generated in the __init__.py file, that contains all the RPC methods as functions but also types that are used in the RPC methods. The nested data structures are generated as TypedDicts.

The types of body, query and params can be obtained from the RPC module using pattern [PascalCaseMethodName][InputType]. For example, a controller method originaly named updateUser will have types UpdateUserBody, UpdateUserQuery and UpdateUserParams generated.


from dist_python.src.vovk_hello_world import UserRPC
import vovk_hello_world
 
def main() -> None:
    body: UserRPC.UpdateUserBody = {
        "email": "john@example.com",
        "profile": {
            "name": "John Doe",
            "age": 25
        }
    }
    query: UserRPC.UpdateUserQuery = {"notify": "email"}
    params: UserRPC.UpdateUserParams = {"id": "123e4567-e89b-12d3-a456-426614174000"}
    # Update user using local module
    update_user_response = UserRPC.update_user(
        params=params,
        body=body,
        query=query
    )
    print('UserRPC.update_user:', update_user_response)
  
if __name__ == "__main__":
  try:
      main()
  except Exception as e:
      print(f"Error: {e}")

Behind the scenes it uses requests for HTTP requests, jsonschema for client-side validation and some other libraries.

JSONLines endpoints

To implement continious streaming with JSONLines endpoints, the client uses Generator to that can be used to iterate over the streamed data.

A controller like this:

StreamController.ts

StreamService.ts

1import { prefix, get, operation } from "vovk";
2import { withZod } from "vovk-zod";
3import { z } from "zod/v4";
4import StreamService from "./StreamService";
5
6@prefix("streams")
7export default class StreamController {
8  @operation({
9    summary: "Stream tokens",
10    description: "Stream tokens to the client",
11  })
12  @get("tokens")
13  static streamTokens = withZod({
14    iteration: z
15      .object({
16        message: z.string().meta({ description: "Message from the token" }),
17      })
18      .meta({
19        description: "Streamed token object",
20      }),
21    async *handle() {
22      yield* StreamService.streamTokens();
23    },
24  });
25}

1import { prefix, get, operation } from "vovk";
2import { withZod } from "vovk-zod";
3import { z } from "zod/v4";
4import StreamService from "./StreamService";
5
6@prefix("streams")
7export default class StreamController {
8  @operation({
9    summary: "Stream tokens",
10    description: "Stream tokens to the client",
11  })
12  @get("tokens")
13  static streamTokens = withZod({
14    iteration: z
15      .object({
16        message: z.string().meta({ description: "Message from the token" }),
17      })
18      .meta({
19        description: "Streamed token object",
20      }),
21    async *handle() {
22      yield* StreamService.streamTokens();
23    },
24  });
25}

1import { prefix, get, operation } from "vovk";
2import { withZod } from "vovk-zod";
3import { z } from "zod/v4";
4import StreamService from "./StreamService";
5
6@prefix("streams")
7export default class StreamController {
8  @operation({
9    summary: "Stream tokens",
10    description: "Stream tokens to the client",
11  })
12  @get("tokens")
13  static streamTokens = withZod({
14    iteration: z
15      .object({
16        message: z.string().meta({ description: "Message from the token" }),
17      })
18      .meta({
19        description: "Streamed token object",
20      }),
21    async *handle() {
22      yield* StreamService.streamTokens();
23    },
24  });
25}

1import { prefix, get, operation } from "vovk";
2import { withZod } from "vovk-zod";
3import { z } from "zod/v4";
4import StreamService from "./StreamService";
5
6@prefix("streams")
7export default class StreamController {
8  @operation({
9    summary: "Stream tokens",
10    description: "Stream tokens to the client",
11  })
12  @get("tokens")
13  static streamTokens = withZod({
14    iteration: z
15      .object({
16        message: z.string().meta({ description: "Message from the token" }),
17      })
18      .meta({
19        description: "Streamed token object",
20      }),
21    async *handle() {
22      yield* StreamService.streamTokens();
23    },
24  });
25}

1import { VovkIteration } from "vovk";
2import type StreamController from "./StreamController";
3
4export default class StreamService {
5  static async *streamTokens() {
6    const tokens: VovkIteration<typeof StreamController.streamTokens>[] =
7      "Vovk.ts is a RESTful back-end meta-framework with RPC built on top of Next.js App router, and this text is a JSONLines stream demo."
8        .match(/[^\s.-]+|\s+|\.|-/g)
9        ?.map((message) => ({ message })) || [];
10
11    for (const token of tokens) {
12      yield token;
13      await new Promise((resolve) => setTimeout(resolve, 100));
14    }
15  }
16}

1import { VovkIteration } from "vovk";
2import type StreamController from "./StreamController";
3
4export default class StreamService {
5  static async *streamTokens() {
6    const tokens: VovkIteration<typeof StreamController.streamTokens>[] =
7      "Vovk.ts is a RESTful back-end meta-framework with RPC built on top of Next.js App router, and this text is a JSONLines stream demo."
8        .match(/[^\s.-]+|\s+|\.|-/g)
9        ?.map((message) => ({ message })) || [];
10
11    for (const token of tokens) {
12      yield token;
13      await new Promise((resolve) => setTimeout(resolve, 100));
14    }
15  }
16}

1import { VovkIteration } from "vovk";
2import type StreamController from "./StreamController";
3
4export default class StreamService {
5  static async *streamTokens() {
6    const tokens: VovkIteration<typeof StreamController.streamTokens>[] =
7      "Vovk.ts is a RESTful back-end meta-framework with RPC built on top of Next.js App router, and this text is a JSONLines stream demo."
8        .match(/[^\s.-]+|\s+|\.|-/g)
9        ?.map((message) => ({ message })) || [];
10
11    for (const token of tokens) {
12      yield token;
13      await new Promise((resolve) => setTimeout(resolve, 100));
14    }
15  }
16}

1import { VovkIteration } from "vovk";
2import type StreamController from "./StreamController";
3
4export default class StreamService {
5  static async *streamTokens() {
6    const tokens: VovkIteration<typeof StreamController.streamTokens>[] =
7      "Vovk.ts is a RESTful back-end meta-framework with RPC built on top of Next.js App router, and this text is a JSONLines stream demo."
8        .match(/[^\s.-]+|\s+|\.|-/g)
9        ?.map((message) => ({ message })) || [];
10
11    for (const token of tokens) {
12      yield token;
13      await new Promise((resolve) => setTimeout(resolve, 100));
14    }
15  }
16}

Will be compiled to the following Python code:

./dist_python/src/package_name/__init__.py


class StreamRPC: 
    # StreamRPC.stream_tokens GET `https://vovk-hello-world.vercel.app/api/streams/tokens`
    class StreamTokensIteration(TypedDict):
        """
        Streamed token object
        """
        message: str
    @staticmethod
    def stream_tokens(
 
        headers: Optional[Dict[str, str]] = None,
        files: Optional[Dict[str, Any]] = None,
        api_root: Optional[str] = None,
        disable_client_validation: bool = False
    ) -> Generator[StreamTokensIteration, None, None]:
        """ 
        Stream tokens
        Description: Stream tokens to the client
        """
        return client.request( # type: ignore
            segment_name='',
            rpc_name='StreamRPC',
            handler_name='streamTokens',
 
            headers=headers,
            files=files,
            api_root=api_root,
            disable_client_validation=disable_client_validation
        )

That can be used like this:


from dist_python.src.vovk_hello_world import StreamRPC  # local module
import vovk_hello_world
 
def main() -> None:
    stream_response = StreamRPC.stream_tokens()
    print("streamTokens:")
    for item in stream_response:
        print(item['message'], end='', flush=True)
 
if __name__ == "__main__":
    try:
        main()
    except Exception as e:
        print(f"Error: {e}")