Command Module

The command module provides comprehensive functionality for creating and managing Neuro API websocket commands. This module handles command formatting, action registration, validation, and type checking for communication with the Neuro Game Interaction API.

Overview

The command module contains all the necessary components to build, format, and validate commands for the Neuro API. It provides functions for creating various types of commands (startup, context, actions, etc.) and includes robust validation for action schemas and TypedDict structures.

Constants

neuro_api.command.ACTION_NAME_ALLOWED_CHARS: Final = frozenset({'-', '0', '1', '2', '3', '4', '5', '6', '7', '8', '9', '_', 'a', 'b', 'c', 'd', 'e', 'f', 'g', 'h', 'i', 'j', 'k', 'l', 'm', 'n', 'o', 'p', 'q', 'r', 's', 't', 'u', 'v', 'w', 'x', 'y', 'z'})

frozenset() -> empty frozenset object frozenset(iterable) -> frozenset object

Build an immutable unordered collection of unique elements.

Frozenset of characters allowed in action names: lowercase letters, digits, underscores, and hyphens.

neuro_api.command.INVALID_SCHEMA_KEYS: Final = frozenset({'$anchor', '$comment', '$defs', '$dynamicAnchor', '$dynamicRef', '$id', '$ref', '$schema', '$vocabulary', 'additionalProperties', 'allOf', 'anyOf', 'contentEncoding', 'contentMediaType', 'contentSchema', 'dependentRequired', 'dependentSchemas', 'deprecated', 'description', 'else', 'if', 'maxProperties', 'minProperties', 'multipleOf', 'not', 'oneOf', 'patternProperties', 'readOnly', 'then', 'title', 'unevaluatedItems', 'unevaluatedProperties', 'writeOnly'})

frozenset() -> empty frozenset object frozenset(iterable) -> frozenset object

Build an immutable unordered collection of unique elements.

Frozenset of JSON Schema keys that are not allowed in action schemas according to the Neuro API specification.

Classes

Action

class neuro_api.command.Action(name: str, description: str, schema: SchemaObject | None = None)

Registerable command that Neuro can execute whenever she wants.

A unique, executable action with well-defined characteristics.

name

A unique identifier for the action. Recommended format: lowercase, with words separated by underscores or dashes (e.g., “join_friend_lobby”, “use_item”).

Type:: str

description

A plain-text description of what the action does. This description will be directly received by Neuro.

Type:: str

schema

A valid JSON schema object describing the expected response data structure. This information will be directly received by Neuro.

Type:: dict[str, object], optional

Notes

If no parameters are needed, omit the schema or set to an empty dict {}.
Schemas must have "type": "object".
For non-object type schemas, wrap the schema in an object with a property.

Examples

>>> action = Action(
...     name="use_item",
...     description="Use an item in the game",
...     schema={"type": "object", "properties": {...}}
... )

A named tuple representing a registerable command that Neuro can execute.

Attributes:

name (str): Unique identifier for the action (lowercase, underscores/hyphens recommended)
description (str): Plain-text description that Neuro will receive
schema (dict[str, object] | None): Optional JSON schema for expected response data

Examples:

# Simple action without parameters
action = Action(
    name="jump",
    description="Make the character jump"
)

# Action with schema for parameters
action = Action(
    name="use_item",
    description="Use an item from inventory",
    schema={
        "type": "object",
        "properties": {
            "item_id": {"type": "string"},
            "quantity": {"type": "integer", "minimum": 1}
        },
        "required": ["item_id"]
    }
)

name: str: Alias for field number 0

description: str: Alias for field number 1

schema: SchemaObject | None: Alias for field number 2

IncomingActionMessageSchema

class neuro_api.command.IncomingActionMessageSchema

Schema for incoming ‘action’ command message fields.

Represents the structure of an action message as specified in the Neuro Game SDK API documentation.

id

Unique identifier for the action message.

Type:: str

name

Name of the action being requested.

Type:: str

data

Additional data associated with the action. This field is not required and can be omitted.

Type:: str, optional

Reference:: Specification details: https://github.com/VedalAI/neuro-game-sdk/blob/main/API/SPECIFICATION.md#action-1

TypedDict schema for incoming ‘action’ command messages from the server.

Attributes:

id (str): Unique identifier for the action message
name (str): Name of the action being requested
data (NotRequired[str]): Optional JSON-stringified data for the action

Command Functions

Client to Server Commands

These functions create commands that clients send to the Neuro server:

neuro_api.command.startup_command(game: str) → bytes

Return formatted startup command.

Client to Server command.

This message should be sent as soon as the game starts, to let Neuro know that the game is running.

This message clears all previously registered actions for this game and does initial setup, and as such should be the very first message that you send.

Parameters:: game (str) – The name of the game to start up.
Returns:: A formatted startup command for the specified game.
Return type:: bytes

Creates the initial startup command that must be sent when a game starts.

# Send startup command for a game
command_bytes = startup_command("My Game")
await client.send_command_data(command_bytes)

neuro_api.command.context_command(game: str, message: str, silent: bool = True) → bytes

Return formatted context command.

Client to Server command.

This message can be sent to let Neuro know about something that is happening in game.

Parameters:

game (str) – The name of the game context is being sent for.
message (str) – A plaintext message that describes what is happening in the game. This information will be directly received by Neuro.
silent (bool, optional) – If True, the message will be added to Neuro’s context without prompting her to respond to it. If False, Neuro _might_ respond to the message directly, unless she is busy talking to someone else or to chat. Defaults to True.

Returns:

A formatted context command for the specified game.

Return type:

bytes

Sends contextual information about game state to Neuro.

# Silent context (won't prompt response)
context_bytes = context_command(
    "My Game",
    "Player entered a new area",
    silent=True
)

# Interactive context (may prompt response)
context_bytes = context_command(
    "My Game",
    "Player is stuck and needs help",
    silent=False
)

neuro_api.command.actions_register_command(game: str, actions: list[Action]) → bytes

Return formatted action/register command.

Client to Server command.

This message registers one or more actions for Neuro to use.

Parameters:

game (str) – The name of the game for which actions are being registered.
actions (list[Action]) – A list of actions to be registered. If you try to register an action that is already registered, it will be ignored.

Returns:

A formatted command to register the specified actions for the game.

Return type:

bytes

Raises:

AssertionError – If the actions list is empty. At least one action must be registered.

Registers one or more actions that Neuro can execute.

actions = [
    Action("move", "Move the player character"),
    Action("attack", "Attack nearby enemies")
]
register_bytes = actions_register_command("My Game", actions)

neuro_api.command.actions_unregister_command(game: str, action_names: Sequence[str]) → bytes

Return formatted action/unregister command.

Client to Server command. This message unregisters one or more actions, preventing Neuro from using them anymore.

Parameters:

game (str) – The name of the game for which actions are being unregistered.
action_names (Sequence[str]) – The names of the actions to unregister. If you try to unregister an action that isn’t registered, there will be no problem.

Returns:

A formatted command to unregister the specified actions for the game.

Return type:

bytes

Raises:

AssertionError – If the action_names sequence is empty. At least one
action name must be provided to unregister. –

Unregisters previously registered actions.

unregister_bytes = actions_unregister_command(
    "My Game",
    ["move", "attack"]
)

neuro_api.command.actions_force_command(game: str, state: str, query: str, action_names: Sequence[str], ephemeral_context: bool = False, priority: ForcePriority = ForcePriority.LOW) → bytes

Return formatted actions/force command.

Client to Server command.

This message forces Neuro to execute one of the listed actions as soon as possible. Note that this might take a bit if she is already talking.

Neuro can only handle one action force at a time. Sending an action force while another one is in progress will cause problems!

Parameters:

game (str) – The name of the game for which actions are being forced.
state (str) – An arbitrary string that describes the current state of the game. This can be plaintext, JSON, Markdown, or any other format. This information will be directly received by Neuro.
query (str) – A plaintext message that tells Neuro what she is currently supposed to be doing (e.g. “It is now your turn. Please perform an action. If you want to use any items, you should use them before picking up the shotgun.”). This information will be directly received by Neuro.
action_names (Sequence[str]) – The names of the actions that Neuro should choose from.
ephemeral_context (bool, optional) – If False, the context provided in the state and query parameters will be remembered by Neuro after the actions force is completed. If True, Neuro will only remember it for the duration of the actions force. Defaults to False.
priority (ForcePriority) – Determines how urgently Neuro should respond to the action force when she is speaking. If Neuro is not speaking, this setting has no effect. The default is ForcePriority.LOW, which will cause Neuro to wait until she finishes speaking before responding. ForcePriority.MEDIUM causes her to finish her current utterance sooner. ForcePriority.HIGH prompts her to process the action force immediately, shortening her utterance and then responding. ForcePriority.CRITICAL will interrupt her speech and make her respond at once. Use ForcePriority.CRITICAL with caution, as it may lead to abrupt and potentially jarring interruptions.

Returns:

A formatted command to force actions for the specified game.

Return type:

bytes

Warning

Neuro can only handle one action force at a time. Sending an action force while another one is in progress will cause problems!

Forces Neuro to execute one of the specified actions immediately.

Warning

Neuro can only handle one action force at a time. Sending multiple forces simultaneously will cause problems.

force_bytes = actions_force_command(
    game="My Game",
    state="Player health: 50%, enemies nearby: 2",
    query="Choose your next action in combat",
    action_names=["attack", "defend", "use_potion"],
    ephemeral_context=True
)

neuro_api.command.actions_result_command(game: str, id_: str, success: bool, message: str | None = None) → bytes

Return formatted action/result command.

Client to Server command.

This message needs to be sent as soon as possible after an action is validated, to allow Neuro to continue.

Until you send an action result, Neuro will just be waiting for the result of her action! Please make sure to send this as soon as possible. It should usually be sent after validating the action parameters, before it is actually executed in-game.

Parameters:

game (str) – The name of the game for which the action result is being reported.
id (str) – The id of the action that this result is for. This is grabbed from the action message directly.
success (bool) – Whether or not the action was successful. _If this is false and this action is part of an actions force, the whole actions force will be immediately retried by Neuro._
message (str, optional) – A plaintext message that describes what happened when the action was executed. If not successful, this should be an error message. If successful, this can either be empty, or provide a _small_ context to Neuro regarding the action she just took (e.g. “Remember to not share this with anyone.”). This information will be directly received by Neuro. Defaults to None.

Returns:

A formatted command to report the result of an action.

Return type:

bytes

Note

Since setting success to false will retry the action force if there was one, if the action was not successful but you don’t want it to be retried, you should set success to true and provide an error message in the message field.

Reports the result of an executed action back to Neuro.

# Successful action
result_bytes = actions_result_command(
    "My Game",
    action_id,
    success=True,
    message="Successfully moved to new location"
)

# Failed action
result_bytes = actions_result_command(
    "My Game",
    action_id,
    success=False,
    message="Cannot move - path is blocked"
)

neuro_api.command.shutdown_ready_command(game: str) → bytes

Return formatted shutdown/ready command.

Client to Server command.

This is part of the game automation API, which will only be used for games that Neuro can launch by herself. As such, most games will not need to implement this.

Parameters:: game (str) – The name of the game that is ready to be shut down.
Returns:: A formatted command indicating the game is ready for shutdown.
Return type:: bytes

Note

This message should be sent as a response to a graceful or an imminent shutdown request, after progress has been saved. After this is sent, Neuro will close the game herself by terminating the process, so to reiterate you must definitely ensure that progress has already been saved.

Warning

This API is only applicable to games that Neuro can launch independently. Most games will not need to use this function.

Indicates the game is ready for shutdown (game automation API only).

Note

This is only used for games that Neuro can launch independently.

Server to Client Commands

These functions create commands that the server sends to clients:

neuro_api.command.action_command(id_: str, name: str, data: str | None = None) → bytes

Return formatted action command.

Server to Client command.

Attempt to execute a registered action.

Parameters:

id (str) – A unique id for the action.
name (str) – The name of the action that Neuro is trying to execute.
data (str) – JSON-stringified data for the action. This _should_ be an object that matches the JSON schema provided when registering the action. If schema was not provided, this should be None.

Returns:

A formatted command to attempt to execute the registered: action.

Return type:

bytes

Server command to execute a registered action.

neuro_api.command.reregister_all_command() → bytes

Return formatted actions/reregister_all command.

Server to Client command.

This signals to the game to unregister all actions and reregister them.

Returns:

A formatted command to unregister all actions and: reregister them.

Return type:

bytes

Warning

This command is part of the proposed API and is not officially supported yet. Some clients may not support it.

Reference:: Specification details: https://github.com/VedalAI/neuro-game-sdk/blob/main/API/PROPOSALS.md#reregister-all-actions

Server command requesting all actions to be unregistered and re-registered.

Warning

This is part of the proposed API and may not be supported by all clients.

neuro_api.command.shutdown_graceful_command(wants_shutdown: bool) → bytes

Return formatted shutdown/graceful command.

Server to Client command.

This message will be sent when Neuro decides to stop playing a game, or upon manual intervention from the dashboard. Signals game to save and quit to main menu and send back a shutdown ready message.

Parameters:

wants_shutdown (bool) – Whether the game should shutdown at the next graceful shutdown point. True means shutdown is requested, False means to cancel the previous shutdown request.

Returns:

A formatted command to request or cancel a graceful: shutdown.

Return type:

bytes

Warning

This is part of the game automation API, which will only be used for games that Neuro can launch by herself. Most games will not support it this command.

Reference:: Specification details: https://github.com/VedalAI/neuro-sdk/blob/main/API/PROPOSALS.md#graceful-shutdown

Server command requesting graceful shutdown.

neuro_api.command.shutdown_immediate_command() → bytes

Return formatted shutdown/immediate command.

Server to Client command.

This signals to the game that it needs to be shutdown immediately and needs to send back a shutdown ready message as soon as the game has saved.

Returns:

A formatted command to inform clients about pending: immediate shutdown.

Return type:

bytes

Warning

This is part of the game automation API, which will only be used for games that Neuro can launch by herself. Most games will not support it this command.

Reference:: Specification details: https://github.com/VedalAI/neuro-sdk/blob/main/API/PROPOSALS.md#immediate-shutdown

Server command requesting immediate shutdown.

Utility Functions

Command Formatting

neuro_api.command.format_command(command: str, game: str | None = None, data: Mapping[str, object] | None = None) → bytes

Return json bytes blob from command details.

Parameters:

command (str) – The websocket command.
game (str | None) – The game name. This is used to identify the game. It should _always_ be the same and should not change. You should use the game’s display name, including any spaces and symbols (e.g. “Buckshot Roulette”). MUST not be present for server messages. Defaults to None.
data (Mapping[str, object], optional) – The command data. This object is different depending on which command you are sending/receiving, and some commands may not have any data, in which case this object will be either undefined or {}. Defaults to None.

Returns:

JSON bytes blob representing the formatted command.

Return type:

bytes

Core function for formatting commands as JSON bytes.

# Server command (no game field)
server_cmd = format_command("some_command", data={"key": "value"})

# Client command (includes game field)
client_cmd = format_command("some_command", "My Game", {"key": "value"})

Type Conversion and Validation

neuro_api.command.convert_parameterized_generic_nonunion(generic: GenericAlias | T) → T | type

Return origin type of aliases.

This function extracts the base type from various generic type aliases, handling different type representations and special cases.

Parameters:

generic (GenericAlias | T) – The generic type or alias to be converted. Can be a GenericAlias or any other type.

Returns:

The origin type of the input generic alias.

For GenericAlias, returns the original type.
For NotRequired types, returns the wrapped type.
For other types, returns the input type as-is.

Return type:

T | type

Note

This function handles special cases like NotRequired from typing and typing_extensions modules.

Extracts origin types from generic aliases, handling NotRequired types.

neuro_api.command.convert_parameterized_generic_union_items(generic: UnionType | T) → T | tuple[type, ...]

Return tuple of based types from union of parameterized generic types.

Parameters:

generic (UnionType | Union | T) – The generic type or alias to be converted. Can be a UnionType or any other type.

Returns:

For UnionType types, returns items as a tuple of types.
For other types, returns the input type as-is.

Return type:

T | tuple[type, …]

Converts union types to tuples of base types.

neuro_api.command.convert_parameterized_generic(generic: GenericAlias | UnionType | T) → T | type | tuple[type, ...]

Return origin type of parameterized generics.

This function extracts the base type from various generic type aliases, handling different type representations and special cases.

Parameters:

generic (GenericAlias | UnionType | T) – The generic type or alias to be converted. Can be a GenericAlias, UnionType, or any other type.

Returns:

The origin type of the input generic alias.

For GenericAlias, returns the original type.
For NotRequired types, returns the wrapped type.
For UnionType types, returns items as a tuple of types.
For other types, returns the input type as-is.

Return type:

T | type | tuple[type, …]

Note

This function handles special cases like NotRequired from typing and typing_extensions modules.

Main function for converting parameterized generic types to their origins.

neuro_api.command.check_typed_dict(data: Mapping[str, object], typed_dict: type[T]) → T

Ensure data matches TypedDict definition and return it as a typed dict.

This function validates that the input data conforms to a TypedDict definition by checking for: - Presence of required keys - Absence of extra keys - Type correctness of values

Parameters:

data (Mapping[str, object]) – The input data to be validated.
typed_dict (type[T]) – The TypedDict class used for validation.

Returns:

The input data converted to the specified TypedDict type.

Return type:

T

Raises:

ValueError – If: - Extra keys are present in the data - Required keys are missing from the data
TypeError – If the type of any key’s value does not match its
annotated type. –

Warning

This function does not work properly for nested types.
Optional keys (NotRequired) are handled, but nested type checking is limited.

Note

The function uses type hints and custom type conversion to perform thorough type checking.

Validates data against TypedDict definitions with comprehensive type checking.

# Validate incoming message data
try:
    validated_action = check_typed_dict(raw_data, IncomingActionMessageSchema)
    action_id = validated_action["id"]
    action_name = validated_action["name"]
except (ValueError, TypeError) as e:
    print(f"Invalid message format: {e}")

Schema Validation

neuro_api.command.check_invalid_keys_recursive(sub_schema: SchemaObject) → list[str]

Recursively checks for invalid keys in the schema.

Parameters:: sub_schema (SchemaObject) – The schema to check for invalid keys.
Returns:: A list of invalid keys that were found.
Return type:: list[str]

Note

Copied from neuro-api-tony/src/neuro_api_tony/api.py found at https://github.com/Pasu4/neuro-api-tony, which is licensed under the MIT License.

Recursively validates JSON schemas for invalid keys.

neuro_api.command.check_action(action: Action) → None

Validate an action before registration.

Performs comprehensive validation of an action to ensure it meets the required naming and schema constraints before being registered.

Parameters:: action (Action) – The action to be validated before registration.
Raises:: ValueError – If: - The action name contains invalid characters - The action schema contains invalid keys

Note

Invalid characters for action names are defined by ACTION_NAME_ALLOWED_CHARS
Schema validation is performed recursively using check_invalid_keys_recursive()

Comprehensive validation of actions before registration.

action = Action(
    name="invalid-name!",  # Contains invalid character
    description="Test action"
)

try:
    check_action(action)
except ValueError as e:
    print(f"Action validation failed: {e}")

Usage Examples

Complete Game Integration

Here’s a complete example showing how to integrate commands in a game client:

from neuro_api.command import *
from neuro_api.client import AbstractNeuroAPIClient

class GameClient(AbstractNeuroAPIClient):
    def __init__(self, websocket, game_name):
        self.websocket = websocket
        self.game_name = game_name
        self.registered_actions = []

    async def write_to_websocket(self, data: str) -> None:
        await self.websocket.send(data)

    async def read_from_websocket(self) -> str:
        return await self.websocket.recv()

    async def initialize_game(self):
        # Send startup command
        startup_cmd = startup_command(self.game_name)
        await self.send_command_data(startup_cmd)

        # Register actions
        actions = [
            Action("move", "Move the player character", {
                "type": "object",
                "properties": {"direction": {"type": "string"}},
                "required": ["direction"]
            }),
            Action("jump", "Make the player jump")
        ]

        # Validate actions before registering
        for action in actions:
            check_action(action)

        register_cmd = actions_register_command(self.game_name, actions)
        await self.send_command_data(register_cmd)
        self.registered_actions = actions

    async def handle_unknown_command(self, command: str, data: dict | None):
        if command == "action":
            await self.handle_action(data)
        else:
            await super().handle_unknown_command(command, data)

    async def handle_action(self, data: dict):
        try:
            action_data = check_typed_dict(data, IncomingActionMessageSchema)
            action_id = action_data["id"]
            action_name = action_data["name"]

            # Execute the action
            success = await self.execute_action(action_name, action_data.get("data"))

            # Send result
            result_cmd = actions_result_command(
                self.game_name,
                action_id,
                success,
                "Action completed successfully" if success else "Action failed"
            )
            await self.send_command_data(result_cmd)

        except (ValueError, TypeError) as e:
            print(f"Invalid action message: {e}")

Error Handling

The module provides comprehensive error handling for various scenarios:

JSON Serialization Errors: orjson.JSONDecodeError when parsing malformed JSON messages.
Type Validation Errors: TypeError when data doesn’t match expected types in TypedDict validation.
Schema Validation Errors: ValueError when actions contain invalid names or schema keys.
Command Format Errors: AssertionError when required parameters are missing (e.g., empty action lists).

try:
    command_bytes = actions_register_command("My Game", [])
except AssertionError:
    print("Must register at least one action")

try:
    check_typed_dict(invalid_data, IncomingActionMessageSchema)
except ValueError as e:
    print(f"Missing required fields: {e}")
except TypeError as e:
    print(f"Type mismatch: {e}")

Best Practices

Action Naming

Use lowercase names with underscores or hyphens: "use_item", "join-lobby"

Schema Design

Always use "type": "object" at the root level, even for simple types:

# Good
schema = {
    "type": "object",
    "properties": {
        "value": {"type": "string"}
    }
}

# Avoid
schema = {"type": "string"}  # Should be wrapped in object

Error Handling

Always validate actions before registration and handle command errors gracefully:

for action in actions:
    try:
        check_action(action)
    except ValueError as e:
        print(f"Skipping invalid action {action.name}: {e}")
        continue

Action Results

Send action results immediately after validation, not after execution:

# Validate parameters first
if not is_valid_move(direction):
    await send_action_result(action_id, False, "Invalid direction")
    return

# Send success result before executing
await send_action_result(action_id, True, "Move validated")

# Then execute the actual move
await execute_move(direction)

API Reference

For complete API specification details, see:

Dependencies

This module requires:

orjson: Fast JSON parsing and serialization
typing_extensions: Extended type hints (NotRequired, is_typeddict)