Skip to content

Message View Classes

Message view classes provide zero-copy access to protocol-specific data structures.

HttpMessage

cymongoose.HttpMessage

Lightweight view over a struct mg_http_message.

Source code in src/cymongoose/_mongoose.pyi 
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
class HttpMessage:
    """Lightweight view over a struct mg_http_message."""

    @property
    def method(self) -> str:
        """HTTP method (GET, POST, etc.)."""
        ...

    @property
    def uri(self) -> str:
        """Request URI path."""
        ...

    @property
    def query(self) -> str:
        """Query string."""
        ...

    @property
    def proto(self) -> str:
        """HTTP protocol version (HTTP/1.1, etc.)."""
        ...

    @property
    def body_bytes(self) -> bytes:
        """Request/response body as bytes."""
        ...

    @property
    def body_text(self) -> str:
        """Request/response body as UTF-8 text."""
        ...

    def header(self, name: str, default: Optional[str] = None) -> Optional[str]:
        """Return a HTTP header value or default when not present.

        Args:
            name: Header name (case-insensitive)
            default: Default value if header not found

        Returns:
            Header value or default
        """
        ...

    def headers(self) -> List[Tuple[str, str]]:
        """Return all HTTP headers as a list of (name, value) tuples."""
        ...

    def query_var(self, name: str) -> Optional[str]:
        """Extract a query string parameter.

        Args:
            name: Parameter name

        Returns:
            Parameter value or None if not found

        Raises:
            ValueError: If decoded value exceeds 2048-byte buffer limit
        """
        ...

    def status(self) -> Optional[int]:
        """Return HTTP status code from response message.

        Returns:
            Status code (e.g., 200, 404) or None if not available
        """
        ...

    def header_var(self, header_name: str, var_name: str) -> Optional[str]:
        """Parse a variable from a header value.

        Useful for parsing sub-values like charset from Content-Type header.
        Example: header_var("Content-Type", "charset") -> "utf-8"

        Args:
            header_name: Name of the header (e.g., "Content-Type")
            var_name: Name of the variable to extract (e.g., "charset")

        Returns:
            The variable value or None if not found
        """
        ...

    def __bool__(self) -> bool: ...

method property

HTTP method (GET, POST, etc.).

uri property

Request URI path.

query property

Query string.

proto property

HTTP protocol version (HTTP/1.1, etc.).

body_bytes property

Request/response body as bytes.

body_text property

Request/response body as UTF-8 text.

header(name, default=None)

Return a HTTP header value or default when not present.

Parameters:

Name Type Description Default
name str

Header name (case-insensitive)

 required
default Optional[str]

Default value if header not found

 None

Returns:

Type Description
Optional[str]

Header value or default
Source code in src/cymongoose/_mongoose.pyi 
86
87
88
89
90
91
92
93
94
95
96
def header(self, name: str, default: Optional[str] = None) -> Optional[str]:
    """Return a HTTP header value or default when not present.

    Args:
        name: Header name (case-insensitive)
        default: Default value if header not found

    Returns:
        Header value or default
    """
    ...

headers()

Return all HTTP headers as a list of (name, value) tuples.

Source code in src/cymongoose/_mongoose.pyi 
 98
 99
100
def headers(self) -> List[Tuple[str, str]]:
    """Return all HTTP headers as a list of (name, value) tuples."""
    ...

query_var(name)

Extract a query string parameter.

Parameters:

Name Type Description Default
name str

Parameter name

 required

Returns:

Type Description
Optional[str]

Parameter value or None if not found

Raises:

Type Description
ValueError

If decoded value exceeds 2048-byte buffer limit
Source code in src/cymongoose/_mongoose.pyi 
102
103
104
105
106
107
108
109
110
111
112
113
114
def query_var(self, name: str) -> Optional[str]:
    """Extract a query string parameter.

    Args:
        name: Parameter name

    Returns:
        Parameter value or None if not found

    Raises:
        ValueError: If decoded value exceeds 2048-byte buffer limit
    """
    ...

status()

Return HTTP status code from response message.

Returns:

Type Description
Optional[int]

Status code (e.g., 200, 404) or None if not available
Source code in src/cymongoose/_mongoose.pyi 
116
117
118
119
120
121
122
def status(self) -> Optional[int]:
    """Return HTTP status code from response message.

    Returns:
        Status code (e.g., 200, 404) or None if not available
    """
    ...

header_var(header_name, var_name)

Parse a variable from a header value.

Useful for parsing sub-values like charset from Content-Type header. Example: header_var("Content-Type", "charset") -> "utf-8"

Parameters:

Name Type Description Default
header_name str

Name of the header (e.g., "Content-Type")

 required
var_name str

Name of the variable to extract (e.g., "charset")

 required

Returns:

Type Description
Optional[str]

The variable value or None if not found
Source code in src/cymongoose/_mongoose.pyi 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
def header_var(self, header_name: str, var_name: str) -> Optional[str]:
    """Parse a variable from a header value.

    Useful for parsing sub-values like charset from Content-Type header.
    Example: header_var("Content-Type", "charset") -> "utf-8"

    Args:
        header_name: Name of the header (e.g., "Content-Type")
        var_name: Name of the variable to extract (e.g., "charset")

    Returns:
        The variable value or None if not found
    """
    ...

__bool__()

Source code in src/cymongoose/_mongoose.pyi 
139
def __bool__(self) -> bool: ...

Overview

HttpMessage provides access to HTTP request/response data without copying. It's a view over the underlying C mg_http_message struct.

Lifetime: Only valid within the event handler for MG_EV_HTTP_MSG or MG_EV_HTTP_HDRS events.

Example

from cymongoose import MG_EV_HTTP_MSG

def handler(conn, ev, data):
    if ev == MG_EV_HTTP_MSG:
        # data is HttpMessage
        print(f"{data.method} {data.uri}")

        # Query parameters
        user_id = data.query_var("id")

        # Headers
        content_type = data.header("Content-Type")
        for name, value in data.headers():
            print(f"{name}: {value}")

        # Body
        body = data.body_bytes  # bytes
        text = data.body_text   # str

        # Status (for responses)
        if data.status():
            print(f"HTTP {data.status()}")

WsMessage

cymongoose.WsMessage

View over an incoming WebSocket frame.

Source code in src/cymongoose/_mongoose.pyi 
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
class WsMessage:
    """View over an incoming WebSocket frame."""

    @property
    def data(self) -> bytes:
        """Frame data as bytes."""
        ...

    @property
    def text(self) -> str:
        """Frame data as UTF-8 text."""
        ...

    @property
    def flags(self) -> int:
        """WebSocket frame flags."""
        ...

data property

Frame data as bytes.

text property

Frame data as UTF-8 text.

flags property

WebSocket frame flags.

Overview

WsMessage provides access to WebSocket frame data.

Lifetime: Only valid within MG_EV_WS_MSG event handler.

Example

from cymongoose import (
    MG_EV_WS_MSG,
    WEBSOCKET_OP_TEXT,
    WEBSOCKET_OP_BINARY,
    WEBSOCKET_OP_PING,
    WEBSOCKET_OP_PONG,
)

def handler(conn, ev, data):
    if ev == MG_EV_WS_MSG:
        # data is WsMessage
        if data.flags == WEBSOCKET_OP_TEXT:
            print(f"Text: {data.text}")
        elif data.flags == WEBSOCKET_OP_BINARY:
            print(f"Binary: {len(data.data)} bytes")
        elif data.flags == WEBSOCKET_OP_PING:
            # Respond to ping
            conn.ws_send(b"", WEBSOCKET_OP_PONG)

MqttMessage

cymongoose.MqttMessage

View over an incoming MQTT message.

Source code in src/cymongoose/_mongoose.pyi 
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
class MqttMessage:
    """View over an incoming MQTT message."""

    @property
    def topic(self) -> str:
        """MQTT topic."""
        ...

    @property
    def data(self) -> bytes:
        """Message payload as bytes."""
        ...

    @property
    def text(self) -> str:
        """Message payload as UTF-8 text."""
        ...

    @property
    def id(self) -> int:
        """Message ID."""
        ...

    @property
    def cmd(self) -> int:
        """MQTT command type."""
        ...

    @property
    def qos(self) -> int:
        """Quality of Service level."""
        ...

    @property
    def ack(self) -> int:
        """Acknowledgement field."""
        ...

    def properties(self) -> List[Dict[str, Any]]:
        """Iterate MQTT v5 properties on this message.

        Returns a list of dicts, each with keys:
            id: property identifier (uint8)
            iv: integer value (for byte/short/int/variable-int types)
            key: string key (only for user-property type)
            val: string value (for string/binary/user-property types)
        """
        ...

topic property

MQTT topic.

data property

Message payload as bytes.

text property

Message payload as UTF-8 text.

id property

Message ID.

cmd property

MQTT command type.

qos property

Quality of Service level.

ack property

Acknowledgement field.

properties()

Iterate MQTT v5 properties on this message.

Returns a list of dicts, each with keys: id: property identifier (uint8) iv: integer value (for byte/short/int/variable-int types) key: string key (only for user-property type) val: string value (for string/binary/user-property types)

Source code in src/cymongoose/_mongoose.pyi 
197
198
199
200
201
202
203
204
205
206
def properties(self) -> List[Dict[str, Any]]:
    """Iterate MQTT v5 properties on this message.

    Returns a list of dicts, each with keys:
        id: property identifier (uint8)
        iv: integer value (for byte/short/int/variable-int types)
        key: string key (only for user-property type)
        val: string value (for string/binary/user-property types)
    """
    ...

Overview

MqttMessage provides access to MQTT message data.

Lifetime: Only valid within MG_EV_MQTT_MSG or MG_EV_MQTT_CMD event handlers.

Example

from cymongoose import MG_EV_MQTT_MSG, MG_EV_MQTT_CMD

def handler(conn, ev, data):
    if ev == MG_EV_MQTT_MSG:
        # data is MqttMessage
        print(f"Topic: {data.topic}")
        print(f"Message: {data.text}")
        print(f"QoS: {data.qos}")
        print(f"Message ID: {data.id}")

        # Parse JSON payload
        import json
        payload = json.loads(data.text)

        # MQTT v5 properties (if broker supports v5)
        for prop in data.properties():
            print(f"Property id={prop['id']}, val={prop['val']}")

    elif ev == MG_EV_MQTT_CMD:
        # MQTT command (CONNECT, SUBSCRIBE, etc.)
        print(f"Command: {data.cmd}")

MQTT v5 Property Type Constants

from cymongoose import (
    MQTT_PROP_TYPE_BYTE,          # 8-bit integer
    MQTT_PROP_TYPE_SHORT,         # 16-bit integer
    MQTT_PROP_TYPE_INT,           # 32-bit integer
    MQTT_PROP_TYPE_VARIABLE_INT,  # Variable-length integer
    MQTT_PROP_TYPE_STRING,        # UTF-8 string
    MQTT_PROP_TYPE_STRING_PAIR,   # Key-value string pair
    MQTT_PROP_TYPE_BINARY_DATA,   # Binary data
)

TlsOpts

cymongoose.TlsOpts

TLS configuration options for secure connections.

Used to configure TLS/SSL settings for HTTPS, WSS, MQTTS, etc.

Source code in src/cymongoose/_mongoose.pyi 
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
class TlsOpts:
    """TLS configuration options for secure connections.

    Used to configure TLS/SSL settings for HTTPS, WSS, MQTTS, etc.
    """

    ca: bytes
    cert: bytes
    key: bytes
    name: bytes
    skip_verification: bool

    def __init__(
        self,
        ca: Union[bytes, str, None] = None,
        cert: Union[bytes, str, None] = None,
        key: Union[bytes, str, None] = None,
        name: Union[bytes, str, None] = None,
        skip_verification: bool = False,
    ) -> None:
        """Initialize TLS options.

        Args:
            ca: CA certificate (PEM format) as bytes or str
            cert: Client certificate (PEM format) as bytes or str
            key: Private key (PEM format) as bytes or str
            name: Server name for SNI (Server Name Indication)
            skip_verification: Skip certificate verification (INSECURE - dev only!)

        Example:
            # Server with custom certificate
            opts = TlsOpts(
                cert=open('server.crt', 'rb').read(),
                key=open('server.key', 'rb').read()
            )

            # Client with custom CA
            opts = TlsOpts(
                ca=open('ca.crt', 'rb').read(),
                name='example.com'
            )

            # Development only - skip verification
            opts = TlsOpts(skip_verification=True)
        """
        ...

ca instance-attribute

cert instance-attribute

key instance-attribute

name instance-attribute

skip_verification instance-attribute

__init__(ca=None, cert=None, key=None, name=None, skip_verification=False)

Initialize TLS options.

Parameters:

Name Type Description Default
ca Union[bytes, str, None]

CA certificate (PEM format) as bytes or str

 None
cert Union[bytes, str, None]

Client certificate (PEM format) as bytes or str

 None
key Union[bytes, str, None]

Private key (PEM format) as bytes or str

 None
name Union[bytes, str, None]

Server name for SNI (Server Name Indication)

 None
skip_verification bool

Skip certificate verification (INSECURE - dev only!)

 False
Example

Server with custom certificate

opts = TlsOpts( cert=open('server.crt', 'rb').read(), key=open('server.key', 'rb').read() )

Client with custom CA

opts = TlsOpts( ca=open('ca.crt', 'rb').read(), name='example.com' )

Development only - skip verification

opts = TlsOpts(skip_verification=True)

Source code in src/cymongoose/_mongoose.pyi 
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
def __init__(
    self,
    ca: Union[bytes, str, None] = None,
    cert: Union[bytes, str, None] = None,
    key: Union[bytes, str, None] = None,
    name: Union[bytes, str, None] = None,
    skip_verification: bool = False,
) -> None:
    """Initialize TLS options.

    Args:
        ca: CA certificate (PEM format) as bytes or str
        cert: Client certificate (PEM format) as bytes or str
        key: Private key (PEM format) as bytes or str
        name: Server name for SNI (Server Name Indication)
        skip_verification: Skip certificate verification (INSECURE - dev only!)

    Example:
        # Server with custom certificate
        opts = TlsOpts(
            cert=open('server.crt', 'rb').read(),
            key=open('server.key', 'rb').read()
        )

        # Client with custom CA
        opts = TlsOpts(
            ca=open('ca.crt', 'rb').read(),
            name='example.com'
        )

        # Development only - skip verification
        opts = TlsOpts(skip_verification=True)
    """
    ...

Overview

TlsOpts configures TLS/SSL settings for secure connections.

Example

from cymongoose import TlsOpts, MG_EV_ACCEPT

# Server with certificate
cert = open("server.crt", "rb").read()
key = open("server.key", "rb").read()

server_opts = TlsOpts(cert=cert, key=key)

# Client with custom CA
ca = open("custom-ca.crt", "rb").read()
client_opts = TlsOpts(ca=ca, name="example.com")

# Development mode (skip verification - INSECURE!)
dev_opts = TlsOpts(skip_verification=True)

def handler(conn, ev, data):
    if ev == MG_EV_ACCEPT:
        # Initialize TLS on new connection
        conn.tls_init(server_opts)

Timer

cymongoose.Timer

Wrapper for Mongoose timer.

Note: The underlying mg_timer is automatically freed by Mongoose when it completes (via MG_TIMER_AUTODELETE flag). This class only manages the Python callback reference.

Source code in src/cymongoose/_mongoose.pyi 
938
939
940
941
942
943
944
945
class Timer:
    """Wrapper for Mongoose timer.

    Note: The underlying mg_timer is automatically freed by Mongoose when it completes
    (via MG_TIMER_AUTODELETE flag). This class only manages the Python callback reference.
    """

    ...

Overview

Timer wraps Mongoose timers. Timers are automatically freed when they complete (MG_TIMER_AUTODELETE flag).

Note

Do not create Timer objects directly. Use Manager.timer_add().

Example

# One-shot timer
timer = manager.timer_add(5000, lambda: print("Hello"))

# Repeating timer
counter = 0
def heartbeat():
    global counter
    counter += 1
    print(f"Heartbeat #{counter}")

timer = manager.timer_add(1000, heartbeat, repeat=True)

# Run immediately and repeat
timer = manager.timer_add(1000, callback, repeat=True, run_now=True)

Event Data Types

Different events provide different data types:

Event Data Type Description
MG_EV_ERROR str Error message
MG_EV_RESOLVE None DNS resolved (check conn.remote_addr)
MG_EV_HTTP_MSG HttpMessage Complete HTTP message
MG_EV_HTTP_HDRS HttpMessage HTTP headers only
MG_EV_WS_OPEN HttpMessage WebSocket handshake request
MG_EV_WS_MSG WsMessage WebSocket frame
MG_EV_MQTT_MSG MqttMessage MQTT message
MG_EV_MQTT_CMD MqttMessage MQTT command
MG_EV_MQTT_OPEN int Connection status code
MG_EV_SNTP_TIME int Milliseconds since epoch
MG_EV_WAKEUP bytes Wakeup payload data
Others None No data

See Also

  • Connection - Connection class
  • Manager - Manager class
  • Guide - Protocol guides