### Install Dependencies

Source: https://github.com/baking-bad/pysignalr/blob/master/CLAUDE.md

Installs project dependencies using uv.

```bash
make install
```

--------------------------------

### Example Usage of EmptyCallback

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/8-types.md

Demonstrates how to use the EmptyCallback for connection open events.

```python
async def on_connected() -> None:
    print('Connected')

client.on_open(on_connected)
```

--------------------------------

### Example Invocation Message Dump

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/8-types.md

Demonstrates creating an InvocationMessage and dumping it to a dictionary.

```python
from pysignalr.messages import InvocationMessage

msg = InvocationMessage('123', 'SendMessage', ['hello'])
dump = msg.dump()
# {'type': 1, 'invocationId': '123', 'target': 'SendMessage', 'arguments': ['hello']}
```

--------------------------------

### Example Usage of AnyCallback

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/8-types.md

Shows how to use AnyCallback for handling incoming messages and request-response scenarios.

```python
async def handle_message(args: list[dict[str, Any]]) -> None:
    print(f'Message: {args}')

async def request_response(args: list[dict[str, Any]]) -> str:
    return 'ack'

client.on('ReceiveMessage', handle_message)
client.on('RequestReply', request_response)
```

--------------------------------

### Install PySignalR

Source: https://github.com/baking-bad/pysignalr/blob/master/README.md

Install the pysignalr library using pip. This is the standard way to add the package to your Python environment.

```bash
pip install pysignalr
```

--------------------------------

### Example Usage of TokenFactory

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/8-types.md

Demonstrates how to define a function to fetch a token and pass it to the SignalRClient constructor via the access_token_factory parameter.

```python
def get_token() -> str:
    # Fetch fresh token from auth service
    return requests.get('https://auth.example.com/token').json()['access_token']

client = SignalRClient(url, access_token_factory=get_token)
```

--------------------------------

### Start Server-to-Client Streaming

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/1-client-signalrclient.md

Initiates a streaming invocation from the server. This method allows receiving multiple items over time, with callbacks for each item, completion, and errors.

```python
async def on_stream_item(item: Any) -> None:
    print(f'Stream item: {item}')

async def on_stream_complete(message: Message) -> None:
    print('Stream complete')

async def on_stream_error(message: CompletionMessage) -> None:
    print(f'Stream error: {message.error}')

await client.stream(
    'GetDataStream',
    ['param1', 'param2'],
    on_next=on_stream_item,
    on_complete=on_stream_complete,
    on_error=on_stream_error
)
```

--------------------------------

### Example Usage of CompletionMessageCallback

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/8-types.md

Demonstrates using CompletionMessageCallback for handling errors reported by the server.

```python
async def handle_error(message: CompletionMessage) -> None:
    print(f'Error: {message.error}')

client.on_error(handle_error)
```

--------------------------------

### Server-to-Client Streaming Example

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/00-START-HERE.md

Use this snippet to stream data from the server to the client. It defines callbacks for receiving individual items, completion messages, and error notifications.

```python
await client.stream(
    'GetData',
    ['param1'],
    on_next=lambda item: print(f'Item: {item}'),
    on_complete=lambda msg: print('Complete'),
    on_error=lambda msg: print(f'Error: {msg.error}'),
)
```

--------------------------------

### Example Usage of MessageCallback

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/8-types.md

Illustrates using MessageCallback to handle invocation responses, including results and errors.

```python
async def on_invocation_response(message: Message) -> None:
    if isinstance(message, CompletionMessage):
        print(f'Result: {message.result}')
        print(f'Error: {message.error}')

await client.send('GetData', [], on_invocation=on_invocation_response)
```

--------------------------------

### Basic SignalR Client Usage

Source: https://github.com/baking-bad/pysignalr/blob/master/README.md

Connect to a SignalR server, subscribe to message handlers, and send initial data. This example connects to the TzKT API and subscribes to operations.

```python
from __future__ import annotations

import asyncio
from contextlib import suppress
from typing import TYPE_CHECKING
from typing import Any

from pysignalr.client import SignalRClient

if TYPE_CHECKING:
    from pysignalr.messages import CompletionMessage


async def on_open() -> None:
    print('Connected to the server')


async def on_close() -> None:
    print('Disconnected from the server')


async def on_message(message: list[dict[str, Any]]) -> None:
    print(f'Received message: {message}')


async def on_client_result(message: list[dict[str, Any]]) -> str:
    """The server can request a result from a client.

    Requires the server to use ISingleClientProxy.InvokeAsync and the client
    to return a result from its .On handler.
    See: https://learn.microsoft.com/en-us/aspnet/core/signalr/hubs?view=aspnetcore-9.0#client-results
    """
    print(f'Received message: {message}')
    return 'reply'


async def on_error(message: CompletionMessage) -> None:
    print(f'Received error: {message.error}')


async def main() -> None:
    client = SignalRClient('https://api.tzkt.io/v1/ws')

    client.on_open(on_open)
    client.on_close(on_close)
    client.on_error(on_error)
    client.on('operations', on_message)
    client.on('client_result', on_client_result)

    await asyncio.gather(
        client.run(),
        client.send('SubscribeToOperations', [{}]),
    )


with suppress(KeyboardInterrupt, asyncio.CancelledError):
    asyncio.run(main())
```

--------------------------------

### run()

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/1-client-signalrclient.md

Starts and maintains the SignalR client connection. This method should be run concurrently with application logic and blocks until the connection is closed or fails.

```APIDOC
## Method: run()

Runs the SignalR client, managing the connection lifecycle. This method blocks until the connection is established and maintained. Call this concurrently with your application logic using `asyncio.gather()` or similar.

### Signature

```python
async def run(self) -> None:
```

**Returns**: None

**Raises**:
- `NegotiationFailure` — If protocol negotiation fails after max retries
- `ServerError` — If the server sends a close message with an error

**Example**:

```python
import asyncio
from pysignalr.client import SignalRClient

client = SignalRClient('https://example.com/chat')
try:
    await client.run()
except Exception as e:
    print(f'Connection failed: {e}')
```
```

--------------------------------

### Run WebsocketTransport

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/5-transport.md

Starts the WebSocket transport, managing the connection lifecycle and automatic reconnection with exponential backoff. Use this to establish and maintain the SignalR connection.

```python
async def run(self) -> None:
```

```python
transport = WebsocketTransport(
    url='https://example.com/chat',
    protocol=JSONProtocol(),
    callback=message_handler
)
try:
    await transport.run()
except NegotiationFailure:
    print('Failed to connect')
```

--------------------------------

### ClientStream.invoke()

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/1-client-signalrclient.md

Sends the InvocationClientStreamMessage to start the stream on the server. This is called automatically by the context manager.

```APIDOC
## ClientStream.invoke()

### Description
Sends the InvocationClientStreamMessage to start the stream on the server. Called automatically by the context manager.

### Method
`invoke(self) -> None`

### Response
#### Success Response
None
```

--------------------------------

### Run SignalR Client Connection

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/1-client-signalrclient.md

Starts and maintains the SignalR client connection. This method should be run concurrently with application logic. It raises exceptions on negotiation failure or server errors.

```python
import asyncio
from pysignalr.client import SignalRClient

client = SignalRClient('https://example.com/chat')
try:
    await client.run()
except Exception as e:
    print(f'Connection failed: {e}')
```

--------------------------------

### Error Handling in PySignalR

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/00-START-HERE.md

This example demonstrates how to handle various exceptions that may occur during client operations, such as authorization failures, connection issues, negotiation problems, and server-side errors.

```python
from pysignalr.exceptions import (
    AuthorizationError,
    ConnectionError,
    NegotiationFailure,
    ServerError
)

try:
    await client.run()
except AuthorizationError:
    print('Authentication failed')
except ConnectionError as e:
    print(f'HTTP {e.status}')
except NegotiationFailure:
    print('Connection timeout')
except ServerError as e:
    print(f'Server error: {e.message}')
```

--------------------------------

### Standard SignalRClient Configuration

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/9-configuration.md

Configures the SignalRClient with a specific URL, JSON protocol, custom headers, and connection timeouts. Useful for typical application setups.

```python
from pysignalr.client import SignalRClient
from pysignalr.protocol.json import JSONProtocol

client = SignalRClient(
    url='https://example.com/api/hub',
    protocol=JSONProtocol(),
    headers={'X-Custom-Header': 'value'},
    connection_timeout=15,
    ping_interval=30,
    signalr_ping_interval=20,
)
```

--------------------------------

### Custom Pickle Protocol Implementation

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/12-internals-advanced.md

Implements a custom SignalR protocol using Python's pickle for serialization and deserialization. This example shows how to extend the base Protocol class to handle custom message formats.

```python
from pysignalr.protocol.abstract import Protocol
from pysignalr.messages import Message, HandshakeRequestMessage, HandshakeResponseMessage
from typing import Iterable
import pickle

class PickleProtocol(Protocol):
    """Custom protocol using Python pickle."""
    
    def __init__(self):
        super().__init__(
            protocol='pickle',
            version=1,
            record_separator=b'\x00'
        )
    
    def decode(self, raw_message: str | bytes) -> Iterable[Message]:
        if isinstance(raw_message, str):
            raw_message = raw_message.encode()
        
        messages = raw_message.split(self.record_separator.encode())
        for msg_bytes in messages:
            if msg_bytes:
                try:
                    msg_dict = pickle.loads(msg_bytes)
                    # Parse dict to Message type...
                    yield self.parse_message(msg_dict)
                except Exception as e:
                    logger.error(f"Failed to decode: {e}")
    
    def encode(self, message: Message | HandshakeRequestMessage) -> bytes:
        data = message.dump()
        return pickle.dumps(data) + self.record_separator
    
    def decode_handshake(self, raw_message: str | bytes) -> tuple[HandshakeResponseMessage, Iterable[Message]]:
        messages = list(self.decode(raw_message))
        handshake_data = messages[0].data if messages else {}
        return HandshakeResponseMessage(handshake_data.get('error')), messages[1:]
    
    @staticmethod
    def parse_message(dict_message):
        # Implementation...
        pass
```

--------------------------------

### Negotiate HTTP POST Response (Azure SignalR)

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/12-internals-advanced.md

Example JSON response from an Azure SignalR negotiate endpoint. It includes the service URL and an access token.

```json
{
  "url": "wss://service.signalr.net/வுகளை",
  "accessToken": "eyJhbGc..."
}
```

--------------------------------

### Implementing a Custom Protocol

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/10-protocol-abstract.md

Define a custom protocol by inheriting from `pysignalr.protocol.abstract.Protocol` and implementing the `decode`, `encode`, and `decode_handshake` methods. This example shows a basic structure for a 'custom' protocol.

```python
from pysignalr.protocol.abstract import Protocol
from pysignalr.messages import (
    Message, HandshakeRequestMessage, HandshakeResponseMessage
)
from typing import Iterable

class CustomProtocol(Protocol):
    def __init__(self):
        super().__init__(
            protocol='custom',
            version=1,
            record_separator=b'\x00'
        )
    
    def decode(self, raw_message: str | bytes) -> Iterable[Message]:
        # Parse raw_message and yield Message objects
        yield from []
    
    def encode(self, message: Message | HandshakeRequestMessage) -> str | bytes:
        # Serialize message
        return b''
    
    def decode_handshake(self, raw_message: str | bytes) -> tuple[HandshakeResponseMessage, Iterable[Message]]:
        # Parse handshake and any following messages
        return HandshakeResponseMessage(error=None), []

# Use custom protocol
client = SignalRClient(
    'https://example.com/hub',
    protocol=CustomProtocol()
)
```

--------------------------------

### Negotiate HTTP POST Response (Standard)

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/12-internals-advanced.md

Example JSON response from a standard SignalR negotiate endpoint. It includes the connection ID, WebSocket URL, and available transports.

```json
{
  "connectionId": "conn-abc123",
  "url": "wss://example.com/api/hub",
  "availableTransports": ["WebSockets"]
}
```

--------------------------------

### Initialize SignalRClient with JSONProtocol

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/10-protocol-abstract.md

Demonstrates how to initialize a SignalRClient using the JSONProtocol. This is the default protocol and can be omitted if not explicitly specified.

```python
from pysignalr.client import SignalRClient
from pysignalr.protocol.json import JSONProtocol

client = SignalRClient(
    'https://example.com/hub',
    protocol=JSONProtocol()  # Default, can be omitted
)
```

--------------------------------

### Run All Commands

Source: https://github.com/baking-bad/pysignalr/blob/master/CLAUDE.md

Executes both linting and testing commands.

```bash
make all
```

--------------------------------

### Import Protocol Implementations

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/README.md

Import JSONProtocol and MessagepackProtocol for selecting the communication protocol.

```python
from pysignalr.protocol.json import JSONProtocol
from pysignalr.protocol.messagepack import MessagepackProtocol
```

--------------------------------

### Initialize SignalRClient with MessagePackProtocol

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/10-protocol-abstract.md

Shows how to initialize a SignalRClient using the MessagepackProtocol. This protocol offers smaller message sizes and faster encoding/decoding compared to JSON.

```python
from pysignalr.client import SignalRClient
from pysignalr.protocol.messagepack import MessagepackProtocol

client = SignalRClient(
    'https://example.com/hub',
    protocol=MessagepackProtocol()
)
```

--------------------------------

### SignalRClient Initialization and Connection Attempt

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/11-init-module.md

This snippet shows how to initialize a SignalRClient and attempt to establish a connection. It includes a try-except block to catch NegotiationFailure, which indicates that all retry attempts have been exhausted.

```python
from pysignalr.client import SignalRClient

client = SignalRClient('https://example.com/hub')

try:
    await client.run()
except NegotiationFailure:
    # Exhausted all retries
    print('Failed to connect after all retries')
```

--------------------------------

### Import Transport and Abstract Components

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/README.md

Import WebsocketTransport and ConnectionState for managing connections, and Protocol for abstract protocol definitions.

```python
from pysignalr.transport.websocket import WebsocketTransport
from pysignalr.transport.abstract import ConnectionState
from pysignalr.protocol.abstract import Protocol
```

--------------------------------

### Basic PySignalR Client Import

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/0-index.md

Import the necessary classes for basic SignalR client usage.

```python
from pysignalr.client import SignalRClient
from pysignalr.messages import CompletionMessage
from pysignalr.exceptions import HubError
```

--------------------------------

### _from_varint()

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/4-protocol-messagepack.md

Decodes a variable-length integer from data starting at a specified offset. This is used internally to read message length prefixes encoded using the varint format.

```APIDOC
## _from_varint()

### Description
Decodes a variable-length integer from data starting at a specified offset. This is used internally to read message length prefixes encoded using the varint format.

### Method
`_from_varint(self, data: bytes, offset: int)`

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Parameters
- **data** (`bytes`) - Required - The data containing the varint
- **offset** (`int`) - Required - The starting offset

### Returns
- **tuple[int, int]** - The decoded integer and the new offset after the varint

**Varint Format**: Each byte has 7 bits of data (bits 0-6) and 1 continuation bit (bit 7). Values continue until a byte with no continuation bit is encountered.

### Example
```python
protocol = MessagepackProtocol()
data = b'\x89\x01more...'  # varint encoding 265
length, next_offset = protocol._from_varint(data, 0)
print(length)      # 265
print(next_offset) # 2
```
```

--------------------------------

### Protocol Constructor

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/10-protocol-abstract.md

Initializes a Protocol object with protocol metadata. Subclasses must inherit from this class.

```python
def __init__(self, protocol: str, version: int, record_separator: str) -> None:
    pass
```

--------------------------------

### on_open()

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/1-client-signalrclient.md

Registers an asynchronous callback function that is invoked once the SignalR connection has been successfully established.

```APIDOC
## Method: on_open()

Registers a callback function to be called when the connection is opened.

### Signature

```python
def on_open(self, callback: EmptyCallback) -> None:
```

**Parameters**

| Parameter | Type | Description |
|-----------|------|-------------|
| callback | `Callable[[], Awaitable[None]]` | Async callable with no parameters |

**Returns**: None

**Example**:

```python
async def connected() -> None:
    print('Connected to server')

client.on_open(connected)
```
```

--------------------------------

### Client-to-Server Streaming Context Manager

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/1-client-signalrclient.md

Provides a context manager for initiating client-to-server streaming. It automatically handles the lifecycle of the stream, including invocation and completion.

```python
async with client.client_stream('UploadData') as stream:
    for item in data:
        await stream.send(item)
# Stream automatically completed on exit
```

--------------------------------

### Basic SignalR Client Connection

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/0-index.md

Establishes a basic connection to a SignalR hub and sets up handlers for connection open and message reception. Requires the `pysignalr.client` module.

```python
from pysignalr.client import SignalRClient

async def main():
    client = SignalRClient('https://example.com/hub')
    
    async def on_connected():
        print('Connected')
    
    async def on_message(args):
        print(f'Message: {args}')
    
    client.on_open(on_connected)
    client.on('ReceiveMessage', on_message)
    
    await client.run()
```

--------------------------------

### Decode Handshake and Subsequent Messages

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/4-protocol-messagepack.md

Decodes a raw message that starts with a JSON-encoded handshake response, followed by MessagePack encoded messages. It separates the handshake response from any subsequent messages.

```python
protocol = MessagepackProtocol()
raw = b'{"error":null}\x1e' + msgpack_encoded_ping
handshake, messages = protocol.decode_handshake(raw)
```

--------------------------------

### Initialize MessagepackProtocol

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/4-protocol-messagepack.md

Initializes a MessagepackProtocol instance. This is used to set up the MessagePack protocol handler for SignalR client communication.

```python
from pysignalr.protocol.messagepack import MessagepackProtocol

protocol = MessagepackProtocol()
client = SignalRClient('https://example.com/chat', protocol=protocol)
```

--------------------------------

### on_open()

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/5-transport.md

Registers a callback function that will be executed when the WebSocket connection is successfully opened.

```APIDOC
## on_open()

### Description
Registers a callback to be called when the connection is opened.

### Method
`def on_open(self, callback: Callable[[], Awaitable[None]]) -> None:`

### Endpoint
N/A (Method call)

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Parameters
- **callback** (`Callable[[], Awaitable[None]]`) - Required - Async callable with no parameters

### Request Example
```python
async def handle_open():
    print('Connection opened!')

transport.on_open(handle_open)
```

### Response
None
```

--------------------------------

### Decode Variable-Length Integer (Varint)

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/4-protocol-messagepack.md

This method decodes a variable-length integer from a byte sequence, starting at a specified offset. It's crucial for reading message length prefixes in the MessagePack protocol. The decoding process continues until a byte without the continuation bit is found.

```python
def _from_varint(self, data: bytes, offset: int) -> tuple[int, int]:
    # ... implementation details omitted for brevity ...
    pass
```

```python
protocol = MessagepackProtocol()
data = b'\x89\x01more...'  # varint encoding 265
length, next_offset = protocol._from_varint(data, 0)
print(length)      # 265
print(next_offset) # 2
```

--------------------------------

### Initialize SignalRClient

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/1-client-signalrclient.md

Instantiate the SignalRClient with the hub URL. This client manages the WebSocket connection and message routing.

```python
from pysignalr.client import SignalRClient

client = SignalRClient('https://example.com/chat')
```

--------------------------------

### Minimal SignalRClient Configuration

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/9-configuration.md

Initializes the SignalRClient with only the required hub URL. Other parameters will use default values.

```python
from pysignalr.client import SignalRClient

client = SignalRClient('https://example.com/hub')
```

--------------------------------

### Server-to-Client Streaming with Callbacks

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/0-index.md

Initiates a server-to-client stream and defines callbacks for receiving individual stream items, completion signals, and errors. The `stream` method is used for this purpose.

```python
async def on_stream_item(item):
    print(f'Item: {item}')

async def on_stream_complete(message):
    print('Stream complete')

async def on_stream_error(message):
    print(f'Error: {message.error}')

await client.stream(
    'GetData',
    ['param1'],
    on_next=on_stream_item,
    on_complete=on_stream_complete,
    on_error=on_stream_error,
)
```

--------------------------------

### run()

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/5-transport.md

Runs the WebSocket transport, managing the connection lifecycle with automatic reconnection. It attempts to establish a connection and handles potential failures.

```APIDOC
## run()

### Description
Runs the WebSocket transport, managing the connection lifecycle with automatic reconnection.

### Method
`async def run(self) -> None:`

### Endpoint
N/A (Method call)

### Parameters
None

### Request Example
```python
transport = WebsocketTransport(
    url='https://example.com/chat',
    protocol=JSONProtocol(),
    callback=message_handler
)
try:
    await transport.run()
except NegotiationFailure:
    print('Failed to connect')
```

### Response
#### Success Response
None (This method manages the connection lifecycle and does not return a value upon successful execution).

#### Error Handling
- **NegotiationFailure**: Raised if negotiation fails after max retries.
```

--------------------------------

### File Organization Structure

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/MANIFEST.md

The project is organized into a clear directory structure, with each file dedicated to a specific logical component of the SignalR client.

```plaintext
0-index.md                      ← Start here
├── 1-client-signalrclient.md   ← Main API
├── 2-messages.md               ← Message types
├── 3-protocol-json.md          ← JSON encoding
├── 4-protocol-messagepack.md   ← MessagePack encoding
├── 5-transport.md              ← WebSocket transport
├── 6-exceptions.md             ← Error handling
├── 7-utilities.md              ← Helper functions
├── 8-types.md                  ← Type reference
├── 9-configuration.md          ← Configuration options
├── 10-protocol-abstract.md     ← Protocol base class
├── 11-init-module.md           ← Package initialization
└── 12-internals-advanced.md    ← Implementation details
README.md                        ← Overview
```

--------------------------------

### Connect with Token Authentication

Source: https://github.com/baking-bad/pysignalr/blob/master/README.md

Use this snippet to connect to a SignalR server requiring token authentication. It demonstrates setting up the client with a URL, an access token factory, and custom headers. The access token factory should contain the logic to fetch or generate your token.

```python
from __future__ import annotations

import asyncio
from contextlib import suppress
from typing import TYPE_CHECKING
from typing import Any

from pysignalr.client import SignalRClient

if TYPE_CHECKING:
    from pysignalr.messages import CompletionMessage


async def on_open() -> None:
    print('Connected to the server')


async def on_close() -> None:
    print('Disconnected from the server')


async def on_message(message: list[dict[str, Any]]) -> None:
    print(f'Received message: {message}')


async def on_client_result(message: list[dict[str, Any]]) -> str:
    """The server can request a result from a client.

    Requires the server to use ISingleClientProxy.InvokeAsync and the client
    to return a result from its .On handler.
    See: https://learn.microsoft.com/en-us/aspnet/core/signalr/hubs?view=aspnetcore-9.0#client-results
    """
    print(f'Received message: {message}')
    return 'reply'


async def on_error(message: CompletionMessage) -> None:
    print(f'Received error: {message.error}')


def token_factory() -> str:
    # Replace with logic to fetch or generate the token
    return 'your_access_token_here'


async def main() -> None:
    client = SignalRClient(
        url='https://api.tzkt.io/v1/ws',
        access_token_factory=token_factory,
        headers={'mycustomheader': 'mycustomheadervalue'},
    )

    client.on_open(on_open)
    client.on_close(on_close)
    client.on_error(on_error)
    client.on('operations', on_message)
    client.on('client_result', on_client_result)

    await asyncio.gather(
        client.run(),
        client.send('SubscribeToOperations', [{}]),
    )


with suppress(KeyboardInterrupt, asyncio.CancelledError):
    asyncio.run(main())

```

--------------------------------

### SignalRClient Constructor

Source: https://github.com/baking-bad/pysignalr/blob/master/README.md

Initializes a new SignalRClient instance to connect to a SignalR server.

```APIDOC
## SignalRClient Constructor

### Description
Initializes a new SignalRClient instance to connect to a SignalR server.

### Parameters
#### Path Parameters
* None

#### Query Parameters
* None

#### Request Body
* None

### Parameters
- **url** (str) - Required - The SignalR server URL
- **protocol** (Protocol | None) - Optional - Protocol for message encoding/decoding (default: JSONProtocol())
- **headers** (dict[str, str] | None) - Optional - Additional headers for the WebSocket handshake
- **ping_interval** (int) - Optional - Keepalive ping interval in seconds (default: 10)
- **connection_timeout** (int) - Optional - Connection timeout in seconds (default: 10)
- **max_size** (int | None) - Optional - Maximum WebSocket message size (default: 1048576)
- **retry_sleep** (float) - Optional - Initial retry delay in seconds (default: 1)
- **retry_multiplier** (float) - Optional - Exponential backoff multiplier (default: 1.1)
- **retry_count** (int) - Optional - Maximum number of retries (default: 10)
- **access_token_factory** (Callable[[], str] | None) - Optional - Function that returns an access token
- **ssl** (ssl.SSLContext | None) - Optional - Custom SSL context

### Methods
- `run()`: Run the client, managing the connection lifecycle.
- `on(event, callback)`: Register a callback for a specific event. If the callback returns a value, it is sent back as a `CompletionMessage` (client results).
- `on_open(callback)`: Register a callback for connection open events.
- `on_close(callback)`: Register a callback for connection close events.
- `on_error(callback)`: Register a callback for error events.
- `send(method, arguments, on_invocation=None)`: Send a message to the server. Without `on_invocation`, sends a non-blocking (fire-and-forget) invocation. With a callback, tracks the invocation and routes the server's completion response to it.
- `stream(event, event_params, on_next=None, on_complete=None, on_error=None)`: Start a server-to-client streaming invocation.
- `client_stream(target)`: Async context manager for client-to-server streaming. Use `await stream.send(item)` inside the context.
```

--------------------------------

### Transport Abstract Methods

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/5-transport.md

Details the abstract methods `run` and `send` that must be implemented by transport subclasses.

```APIDOC
## Abstract Base Class: Transport

Abstract base class for implementing a transport protocol.

### Attributes

| Attribute | Type | Description |
|-----------|------|-------------|
| protocol | `Protocol` | The protocol used by the transport |
| state | `ConnectionState` | The current state of the connection |

### Abstract Methods

#### run()

```python
async def run(self) -> None:
```

Abstract method for running the transport protocol. Must be implemented by subclasses.

#### send()

```python
async def send(self, message: Message) -> None:
```

Abstract method for sending a message. Must be implemented by subclasses.

**Parameters**

| Parameter | Type | Description |
|-----------|------|-------------|
| message | `Message` | The message to be sent |
```

--------------------------------

### BaseJSONProtocol Constructor

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/3-protocol-json.md

Initializes the BaseJSONProtocol. This class is generally not used directly.

```python
def __init__(self) -> None:
```

--------------------------------

### Construct WebSocket URL with Connection ID

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/12-internals-advanced.md

Demonstrates the process of constructing a WebSocket URL for a SignalR connection after negotiation. It involves appending the connection ID and converting the scheme.

```text
https://example.com/api/hub + /negotiate
↓ (get connectionId)
https://example.com/api/hub?id=conn-abc123
↓ (convert scheme)
wss://example.com/api/hub?id=conn-abc123
```

--------------------------------

### Run Tests

Source: https://github.com/baking-bad/pysignalr/blob/master/CLAUDE.md

Executes pytest with coverage reporting.

```bash
make test
```

--------------------------------

### Construct Negotiation URL

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/7-utilities.md

Constructs the negotiation URL by appending '/negotiate' and converting the scheme to HTTP. This is used internally to obtain a connection ID.

```python
from pysignalr.utils import get_negotiate_url

# WebSocket URL
url1 = get_negotiate_url('wss://example.com/chat')
print(url1)  # 'https://example.com/chat/negotiate'

# HTTP URL
url2 = get_negotiate_url('https://example.com/hub')
print(url2)  # 'https://example.com/hub/negotiate'

# With path and query string
url3 = get_negotiate_url('wss://example.com:8080/api/chat?token=abc')
print(url3)  # 'https://example.com:8080/api/chat/negotiate?token=abc'
```

--------------------------------

### client_stream()

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/1-client-signalrclient.md

A context manager for handling client-to-server streaming, automatically managing the stream's lifecycle.

```APIDOC
## client_stream()

### Description
Context manager for client-to-server streaming. Automatically handles stream lifecycle (invoke on enter, complete on exit).

### Parameters
#### Path Parameters
* None

#### Query Parameters
* None

#### Request Body
* None

### Method Signature
```python
@asynccontextmanager
async def client_stream(self, target: str) -> AsyncIterator[ClientStream]:
```

### Parameters
* **target** (`str`) - Required - The target method name on the server.

### Yields
* `ClientStream` — The client stream instance.

### Returns
None (context manager)

### Example
```python
async with client.client_stream('UploadData') as stream:
    for item in data:
        await stream.send(item)
# Stream automatically completed on exit
```
```

--------------------------------

### Client-to-Server Streaming

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/README.md

Enables streaming data from the client to the server. Use the `client_stream` context manager to send multiple items.

```python
async with client.client_stream('UploadData') as stream:
    for item in items:
        await stream.send(item)
```

--------------------------------

### Initialize JSONProtocol

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/3-protocol-json.md

Initializes a JSONProtocol instance. This class handles SignalR's JSON message encoding and decoding using the record separator character.

```python
from pysignalr.protocol.json import JSONProtocol

protocol = JSONProtocol()
client = SignalRClient('https://example.com/chat', protocol=protocol)
```

--------------------------------

### Register Connection Open Callback

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/1-client-signalrclient.md

Register an asynchronous callback function to be executed once the SignalR connection is successfully established.

```python
async def connected() -> None:
    print('Connected to server')

client.on_open(connected)
```

--------------------------------

### Run Linting

Source: https://github.com/baking-bad/pysignalr/blob/master/CLAUDE.md

Runs code quality checks including black, ruff, and mypy.

```bash
make lint
```

--------------------------------

### stream()

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/1-client-signalrclient.md

Initiates a server-to-client streaming invocation, allowing for multiple `StreamItemMessage` responses followed by a `CompletionMessage`.

```APIDOC
## stream()

### Description
Starts a server-to-client streaming invocation. The server responds with zero or more `StreamItemMessage` routed to `on_next`, followed by a `CompletionMessage` routed to `on_complete`.

### Parameters
#### Path Parameters
* None

#### Query Parameters
* None

#### Request Body
* None

### Method Signature
```python
async def stream(
    self,
    event: str,
    event_params: list[str],
    on_next: MessageCallback | None = None,
    on_complete: MessageCallback | None = None,
    on_error: CompletionMessageCallback | None = None,
) -> None:
```

### Parameters
* **event** (`str`) - Required - The hub method name to stream from.
* **event_params** (`list[str]`) - Required - The arguments for the streaming method.
* **on_next** (`Callable[[Message], Awaitable[None]] | None`) - Optional - Called with each stream item's payload.
* **on_complete** (`Callable[[Message], Awaitable[None]] | None`) - Optional - Called on the final `CompletionMessage`.
* **on_error** (`Callable[[CompletionMessage], Awaitable[None]] | None`) - Optional - Called on error; falls back to global `on_error`.

### Returns
None

### Raises
* `RuntimeError` — If `on_error` callback is not set and an error occurs

### Example
```python
async def on_stream_item(item: Any) -> None:
    print(f'Stream item: {item}')

async def on_stream_complete(message: Message) -> None:
    print('Stream complete')

async def on_stream_error(message: CompletionMessage) -> None:
    print(f'Stream error: {message.error}')

await client.stream(
    'GetDataStream',
    ['param1', 'param2'],
    on_next=on_stream_item,
    on_complete=on_stream_complete,
    on_error=on_stream_error
)
```
```

--------------------------------

### SignalRClient Configuration with Token Authentication

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/9-configuration.md

Sets up SignalRClient for token-based authentication by providing a function to dynamically fetch an access token. The token is sent as an Authorization: Bearer header.

```python
from pysignalr.client import SignalRClient

def get_token() -> str:
    # Fetch token from auth service or cache
    return 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...'

client = SignalRClient(
    url='https://example.com/api/hub',
    access_token_factory=get_token,
)
```

--------------------------------

### Catching PySignalR Exceptions

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/6-exceptions.md

Demonstrates how to use a try-except block to catch and handle specific exceptions raised by the PySignalR client during its operation. This includes errors during connection, negotiation, and server communication.

```python
from pysignalr.client import SignalRClient
from pysignalr.exceptions import (
    AuthorizationError,
    ConnectionError,
    ServerError,
    NegotiationFailure,
    HubError
)

async def main():
    client = SignalRClient('https://example.com/hub')
    
    try:
        await client.run()
    except AuthorizationError:
        print('Invalid credentials')
    except ConnectionError as e:
        print(f'HTTP {e.status}')
    except NegotiationFailure:
        print('Cannot connect after retries')
    except ServerError as e:
        print(f'Server error: {e.message}')
    except HubError as e:
        print(f'Unknown hub error: {e}')
```

--------------------------------

### Real-time Dashboard Configuration

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/9-configuration.md

Configure a SignalRClient to connect to the TzKT API for real-time dashboard updates. Sets custom intervals for ping and connection timeouts.

```python
from pysignalr.client import SignalRClient

client = SignalRClient(
    'https://api.tzkt.io/v1/ws',
    ping_interval=30,
    signalr_ping_interval=25,
    connection_timeout=15,
)
```

--------------------------------

### Pysignalr File Structure

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/README.md

Overview of the directory and file layout for the pysignalr library. Shows the organization of modules for client, messages, utilities, protocols, and transports.

```text
pysignalr/
├── __init__.py                    # Backoff initialization
├── client.py                      # SignalRClient, ClientStream
├── messages.py                    # Message types
├── exceptions.py                  # Exception classes
├── utils.py                       # URL utilities
├── protocol/
│   ├── __init__.py
│   ├── abstract.py                # Protocol base class
│   ├── json.py                    # JSONProtocol
│   └── messagepack.py             # MessagepackProtocol
└── transport/
    ├── __init__.py
    ├── abstract.py                # Transport base, ConnectionState
    └── websocket.py               # WebsocketTransport
```

--------------------------------

### Method Invocation with Response Handling

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/0-index.md

Sends a method invocation to the server and specifies a callback to handle the server's response, including the result or any errors. The `send` method is used, and the response is processed in `on_invocation`.

```python
async def on_response(message):
    if isinstance(message, CompletionMessage):
        print(f'Result: {message.result}')

await client.send(
    'GetTime',
    [],
    on_invocation=on_response,
)
```

--------------------------------

### Basic SignalR Connection

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/00-START-HERE.md

Establishes a basic connection to a SignalR hub. Use this for simple client-server communication without authentication.

```python
from pysignalr.client import SignalRClient

client = SignalRClient('https://example.com/hub')

client.on_open(lambda: print('Connected'))
client.on('MessageReceived', lambda args: print(f'Message: {args}'))
client.on_error(lambda msg: print(f'Error: {msg.error}'))

await client.run()
```

--------------------------------

### WebsocketTransport Constructor

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/5-transport.md

Initializes the WebSocket transport with various configuration options including URL, protocol, callbacks, and connection parameters.

```python
def __init__(
    self,
    url: str,
    protocol: Protocol,
    callback: Callable[[Message], Awaitable[None]],
    headers: dict[str, str] | None = None,
    skip_negotiation: bool = False,
    ping_interval: int = DEFAULT_PING_INTERVAL,
    signalr_ping_interval: int = DEFAULT_SIGNALR_PING_INTERVAL,
    connection_timeout: int = DEFAULT_CONNECTION_TIMEOUT,
    retry_sleep: float = DEFAULT_RETRY_SLEEP,
    retry_multiplier: float = DEFAULT_RETRY_MULTIPLIER,
    retry_count: int = DEFAULT_RETRY_COUNT,
    max_size: int | None = DEFAULT_MAX_SIZE,
    access_token_factory: Callable[[], str] | None = None,
    ssl: ssl.SSLContext | None = None,
) -> None:
```

--------------------------------

### PySignalR All Message Type Imports

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/0-index.md

Import all available message types for comprehensive message handling.

```python
from pysignalr.messages import (
    Message, MessageType,
    InvocationMessage, StreamInvocationMessage,
    StreamItemMessage, CompletionMessage,
    CancelInvocationMessage, PingMessage, CloseMessage,
    CompletionClientStreamMessage, InvocationClientStreamMessage,
)
```

--------------------------------

### send()

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/1-client-signalrclient.md

Invokes a specified hub method on the server. This can be a fire-and-forget call or include a callback to handle the server's completion response.

```APIDOC
## send()

### Description
Invokes a hub method on the server. Without `on_invocation` this is a fire-and-forget call. With a callback, an `invocationId` is generated and the server's `CompletionMessage` response is routed to the callback.

### Parameters
#### Path Parameters
* None

#### Query Parameters
* None

#### Request Body
* None

### Method Signature
```python
async def send(
    self,
    method: str,
    arguments: list[dict[str, Any]],
    on_invocation: MessageCallback | None = None,
) -> None:
```

### Parameters
* **method** (`str`) - Required - The hub method name to invoke.
* **arguments** (`list[dict[str, Any]]`) - Required - The arguments to pass to the method.
* **on_invocation** (`Callable[[Message], Awaitable[None]] | None`) - Optional - Callback for the completion response.

### Returns
None

### Raises
* `RuntimeError` — If trying to send before the connection is established

### Example
```python
# Fire-and-forget (no response expected)
await client.send('SendMessage', [{'user': 'Alice', 'message': 'Hello'}])

# With response callback
async def on_response(message: Message) -> None:
    if isinstance(message, CompletionMessage):
        print(f'Server result: {message.result}')

await client.send(
    'GetServerTime',
    [{}],
    on_invocation=on_response
)
```
```

--------------------------------

### PySignalR Custom Protocol Imports

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/0-index.md

Import base protocol and specific protocol implementations for custom protocol handling.

```python
from pysignalr.protocol.abstract import Protocol
from pysignalr.protocol.json import JSONProtocol
from pysignalr.protocol.messagepack import MessagepackProtocol
```

--------------------------------

### on()

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/1-client-signalrclient.md

Registers an asynchronous callback function to be executed when a specific hub method is invoked by the server. Supports client-side result reporting.

```APIDOC
## Method: on()

Registers a callback for a hub method invocation. If the callback returns a value and the invocation has an `invocation_id`, the result is sent back to the server as a `CompletionMessage` (client results).

### Signature

```python
def on(self, event: str, callback: AnyCallback) -> None:
```

**Parameters**

| Parameter | Type | Description |
|-----------|------|-------------|
| event | `str` | The hub method name to listen for |
| callback | `Callable[[Any], Awaitable[Any \| None]]` | Async callable invoked with the message arguments list |

**Returns**: None

**Example**:

```python
async def on_message(args: list[dict[str, Any]]) -> None:
    print(f'Received: {args}')

async def on_request(args: list[dict[str, Any]]) -> str:
    # Return a value to send back to the server (client results)
    return 'acknowledgment'

client.on('MessageReceived', on_message)
client.on('RequestFromServer', on_request)
```
```

--------------------------------

### Sending Items with ClientStream

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/1-client-signalrclient.md

Use the `send` method within a `client_stream` context manager to send individual items to the server. This is suitable for streaming multiple data points to a single server operation.

```python
async with client.client_stream('ProcessItems') as stream:
    await stream.send({'id': 1, 'value': 'data1'})
    await stream.send({'id': 2, 'value': 'data2'})
```

--------------------------------

### Handle SignalR Errors with PySignalR

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/0-index.md

This snippet demonstrates how to set up an error handler for a SignalR client and catch various exceptions that may occur during connection and operation. It covers authorization, connection, negotiation, server, and hub-specific errors.

```python
from pysignalr.client import SignalRClient
from pysignalr.exceptions import (
    AuthorizationError,
    ConnectionError,
    NegotiationFailure,
    ServerError,
    HubError,
)

async def main():
    client = SignalRClient('https://example.com/hub')
    
    async def on_error(message):
        print(f'Server error: {message.error}')
    
    client.on_error(on_error)
    
    try:
        await client.run()
    except AuthorizationError:
        print('Check credentials')
    except ConnectionError as e:
        print(f'HTTP {e.status}')
    except NegotiationFailure:
        print('Negotiation failed after retries')
    except ServerError as e:
        print(f'Server error: {e.message}')
    except HubError as e:
        print(f'Unexpected error: {e}')
```

--------------------------------

### Client-to-Server Streaming

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/0-index.md

Enables client-to-server streaming by using an asynchronous context manager. Data items are sent iteratively within the `with` block using the `stream.send()` method.

```python
async with client.client_stream('UploadData') as stream:
    for item in data:
        await stream.send(item)
```

--------------------------------

### Register OnOpen Callback for WebsocketTransport

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/5-transport.md

Registers an asynchronous callback function to be executed when the WebSocket connection is successfully opened. The callback takes no arguments.

```python
def on_open(self, callback: Callable[[], Awaitable[None]]) -> None:
```

--------------------------------

### Handle AuthorizationError Exception

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/6-exceptions.md

Shows how to catch an AuthorizationError, which is raised when authentication fails during the client's negotiation phase.

```python
from pysignalr.exceptions import AuthorizationError

try:
    await client.run()
except AuthorizationError:
    print('Authentication failed')
```

--------------------------------

### Run Single Test

Source: https://github.com/baking-bad/pysignalr/blob/master/CLAUDE.md

Executes a specific test file using pytest with asyncio mode.

```bash
pytest --asyncio-mode=auto -s -v tests/test_pysignalr/test_pysignalr.py::test_name
```

--------------------------------

### Method Invocation with Response Handler

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/README.md

Invokes a server method and specifies a callback function to handle the response. Useful for requests expecting a single result.

```python
async def on_response(message):
    print(f'Result: {message.result}')

await client.send('GetTime', [], on_invocation=on_response)
```

--------------------------------

### Construct Connection URL

Source: https://github.com/baking-bad/pysignalr/blob/master/_autodocs/7-utilities.md

Constructs the final WebSocket connection URL, including the connection ID as a query parameter and ensuring a WebSocket scheme. Handles existing query strings.

```python
from pysignalr.utils import get_connection_url

# From HTTP endpoint
url1 = get_connection_url('https://example.com/chat', 'conn-12345')
print(url1)  # 'wss://example.com/chat?id=conn-12345'

# With existing query string
url2 = get_connection_url('https://example.com/hub?token=xyz', 'conn-67890')
print(url2)  # 'wss://example.com/hub?token=xyz&id=conn-67890'

# Already WebSocket URL (scheme unchanged)
url3 = get_connection_url('wss://example.com/chat', 'conn-abc')
print(url3)  # 'wss://example.com/chat?id=conn-abc'
```