Skip to content

Client API Reference

This page documents the main client classes and functions in kalshi-py.

Client Classes

kalshi_py.client.Client

A class for keeping track of data related to the API

The following are accepted as keyword arguments and will be used to construct httpx Clients internally:

``base_url``: The base URL for the API, all requests are made to a relative path to this URL

``cookies``: A dictionary of cookies to be sent with every request

``headers``: A dictionary of headers to be sent with every request

``timeout``: The maximum amount of a time a request can take. API functions will raise
httpx.TimeoutException if this is exceeded.

``verify_ssl``: Whether or not to verify the SSL certificate of the API server. This should be True in production,
but can be set to False for testing purposes.

``follow_redirects``: Whether or not to follow redirects. Default value is False.

``httpx_args``: A dictionary of additional arguments to be passed to the ``httpx.Client`` and ``httpx.AsyncClient`` constructor.

Attributes:

Name Type Description
raise_on_unexpected_status bool

Whether or not to raise an errors.UnexpectedStatus if the API returns a status code that was not documented in the source OpenAPI document. Can also be provided as a keyword argument to the constructor.

Functions

with_headers(headers: dict[str, str]) -> Client

Get a new client matching this one with additional headers

Source code in kalshi_py/client.py 
48
49
50
51
52
53
54
def with_headers(self, headers: dict[str, str]) -> "Client":
    """Get a new client matching this one with additional headers"""
    if self._client is not None:
        self._client.headers.update(headers)
    if self._async_client is not None:
        self._async_client.headers.update(headers)
    return evolve(self, headers={**self._headers, **headers})

with_cookies(cookies: dict[str, str]) -> Client

Get a new client matching this one with additional cookies

Source code in kalshi_py/client.py 
56
57
58
59
60
61
62
def with_cookies(self, cookies: dict[str, str]) -> "Client":
    """Get a new client matching this one with additional cookies"""
    if self._client is not None:
        self._client.cookies.update(cookies)
    if self._async_client is not None:
        self._async_client.cookies.update(cookies)
    return evolve(self, cookies={**self._cookies, **cookies})

with_timeout(timeout: httpx.Timeout) -> Client

Get a new client matching this one with a new timeout (in seconds)

Source code in kalshi_py/client.py 
64
65
66
67
68
69
70
def with_timeout(self, timeout: httpx.Timeout) -> "Client":
    """Get a new client matching this one with a new timeout (in seconds)"""
    if self._client is not None:
        self._client.timeout = timeout
    if self._async_client is not None:
        self._async_client.timeout = timeout
    return evolve(self, timeout=timeout)

set_httpx_client(client: httpx.Client) -> Client

Manually set the underlying httpx.Client

NOTE: This will override any other settings on the client, including cookies, headers, and timeout.

Source code in kalshi_py/client.py 
72
73
74
75
76
77
78
def set_httpx_client(self, client: httpx.Client) -> "Client":
    """Manually set the underlying httpx.Client

    **NOTE**: This will override any other settings on the client, including cookies, headers, and timeout.
    """
    self._client = client
    return self

get_httpx_client() -> httpx.Client

Get the underlying httpx.Client, constructing a new one if not previously set

Source code in kalshi_py/client.py 
80
81
82
83
84
85
86
87
88
89
90
91
92
def get_httpx_client(self) -> httpx.Client:
    """Get the underlying httpx.Client, constructing a new one if not previously set"""
    if self._client is None:
        self._client = httpx.Client(
            base_url=self._base_url,
            cookies=self._cookies,
            headers=self._headers,
            timeout=self._timeout,
            verify=self._verify_ssl,
            follow_redirects=self._follow_redirects,
            **self._httpx_args,
        )
    return self._client

set_async_httpx_client(async_client: httpx.AsyncClient) -> Client

Manually the underlying httpx.AsyncClient

NOTE: This will override any other settings on the client, including cookies, headers, and timeout.

Source code in kalshi_py/client.py 
103
104
105
106
107
108
109
def set_async_httpx_client(self, async_client: httpx.AsyncClient) -> "Client":
    """Manually the underlying httpx.AsyncClient

    **NOTE**: This will override any other settings on the client, including cookies, headers, and timeout.
    """
    self._async_client = async_client
    return self

get_async_httpx_client() -> httpx.AsyncClient

Get the underlying httpx.AsyncClient, constructing a new one if not previously set

Source code in kalshi_py/client.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
def get_async_httpx_client(self) -> httpx.AsyncClient:
    """Get the underlying httpx.AsyncClient, constructing a new one if not previously set"""
    if self._async_client is None:
        self._async_client = httpx.AsyncClient(
            base_url=self._base_url,
            cookies=self._cookies,
            headers=self._headers,
            timeout=self._timeout,
            verify=self._verify_ssl,
            follow_redirects=self._follow_redirects,
            **self._httpx_args,
        )
    return self._async_client

handler: python options: show_source: true show_root_heading: true

kalshi_py.client.AuthenticatedClient

A Client which has been authenticated for use on secured endpoints

The following are accepted as keyword arguments and will be used to construct httpx Clients internally:

``base_url``: The base URL for the API, all requests are made to a relative path to this URL

``cookies``: A dictionary of cookies to be sent with every request

``headers``: A dictionary of headers to be sent with every request

``timeout``: The maximum amount of a time a request can take. API functions will raise
httpx.TimeoutException if this is exceeded.

``verify_ssl``: Whether or not to verify the SSL certificate of the API server. This should be True in production,
but can be set to False for testing purposes.

``follow_redirects``: Whether or not to follow redirects. Default value is False.

``httpx_args``: A dictionary of additional arguments to be passed to the ``httpx.Client`` and ``httpx.AsyncClient`` constructor.

Attributes:

Name Type Description
raise_on_unexpected_status bool

Whether or not to raise an errors.UnexpectedStatus if the API returns a status code that was not documented in the source OpenAPI document. Can also be provided as a keyword argument to the constructor.
token str

The token to use for authentication
prefix str

The prefix to use for the Authorization header
auth_header_name str

The name of the Authorization header

Functions

with_headers(headers: dict[str, str]) -> AuthenticatedClient

Get a new client matching this one with additional headers

Source code in kalshi_py/client.py 
182
183
184
185
186
187
188
def with_headers(self, headers: dict[str, str]) -> "AuthenticatedClient":
    """Get a new client matching this one with additional headers"""
    if self._client is not None:
        self._client.headers.update(headers)
    if self._async_client is not None:
        self._async_client.headers.update(headers)
    return evolve(self, headers={**self._headers, **headers})

with_cookies(cookies: dict[str, str]) -> AuthenticatedClient

Get a new client matching this one with additional cookies

Source code in kalshi_py/client.py 
190
191
192
193
194
195
196
def with_cookies(self, cookies: dict[str, str]) -> "AuthenticatedClient":
    """Get a new client matching this one with additional cookies"""
    if self._client is not None:
        self._client.cookies.update(cookies)
    if self._async_client is not None:
        self._async_client.cookies.update(cookies)
    return evolve(self, cookies={**self._cookies, **cookies})

with_timeout(timeout: httpx.Timeout) -> AuthenticatedClient

Get a new client matching this one with a new timeout (in seconds)

Source code in kalshi_py/client.py 
198
199
200
201
202
203
204
def with_timeout(self, timeout: httpx.Timeout) -> "AuthenticatedClient":
    """Get a new client matching this one with a new timeout (in seconds)"""
    if self._client is not None:
        self._client.timeout = timeout
    if self._async_client is not None:
        self._async_client.timeout = timeout
    return evolve(self, timeout=timeout)

set_httpx_client(client: httpx.Client) -> AuthenticatedClient

Manually set the underlying httpx.Client

NOTE: This will override any other settings on the client, including cookies, headers, and timeout.

Source code in kalshi_py/client.py 
206
207
208
209
210
211
212
def set_httpx_client(self, client: httpx.Client) -> "AuthenticatedClient":
    """Manually set the underlying httpx.Client

    **NOTE**: This will override any other settings on the client, including cookies, headers, and timeout.
    """
    self._client = client
    return self

get_httpx_client() -> httpx.Client

Get the underlying httpx.Client, constructing a new one if not previously set

Source code in kalshi_py/client.py 
214
215
216
217
218
219
220
221
222
223
224
225
226
227
def get_httpx_client(self) -> httpx.Client:
    """Get the underlying httpx.Client, constructing a new one if not previously set"""
    if self._client is None:
        self._headers[self.auth_header_name] = f"{self.prefix} {self.token}" if self.prefix else self.token
        self._client = httpx.Client(
            base_url=self._base_url,
            cookies=self._cookies,
            headers=self._headers,
            timeout=self._timeout,
            verify=self._verify_ssl,
            follow_redirects=self._follow_redirects,
            **self._httpx_args,
        )
    return self._client

set_async_httpx_client(async_client: httpx.AsyncClient) -> AuthenticatedClient

Manually the underlying httpx.AsyncClient

NOTE: This will override any other settings on the client, including cookies, headers, and timeout.

Source code in kalshi_py/client.py 
238
239
240
241
242
243
244
def set_async_httpx_client(self, async_client: httpx.AsyncClient) -> "AuthenticatedClient":
    """Manually the underlying httpx.AsyncClient

    **NOTE**: This will override any other settings on the client, including cookies, headers, and timeout.
    """
    self._async_client = async_client
    return self

get_async_httpx_client() -> httpx.AsyncClient

Get the underlying httpx.AsyncClient, constructing a new one if not previously set

Source code in kalshi_py/client.py 
246
247
248
249
250
251
252
253
254
255
256
257
258
259
def get_async_httpx_client(self) -> httpx.AsyncClient:
    """Get the underlying httpx.AsyncClient, constructing a new one if not previously set"""
    if self._async_client is None:
        self._headers[self.auth_header_name] = f"{self.prefix} {self.token}" if self.prefix else self.token
        self._async_client = httpx.AsyncClient(
            base_url=self._base_url,
            cookies=self._cookies,
            headers=self._headers,
            timeout=self._timeout,
            verify=self._verify_ssl,
            follow_redirects=self._follow_redirects,
            **self._httpx_args,
        )
    return self._async_client

handler: python options: show_source: true show_root_heading: true

kalshi_py.auth.KalshiAuthenticatedClient(access_key_id: str, private_key_pem: str, base_url: str = 'https://api.elections.kalshi.com/trade-api/v2', **kwargs: Any)

Bases: AuthenticatedClient

Authenticated client for Kalshi API with RSA-PSS signature authentication.

Initialize authenticated client.

Parameters:

Name Type Description Default
access_key_id str

Kalshi access key ID (required)

 required
private_key_pem str

PEM-encoded private key (required)

 required
base_url str

API base URL (defaults to elections API)

 'https://api.elections.kalshi.com/trade-api/v2'
**kwargs Any

Additional arguments passed to base Client

 {}
Source code in kalshi_py/auth.py 
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
def __init__(
    self,
    access_key_id: str,
    private_key_pem: str,
    base_url: str = "https://api.elections.kalshi.com/trade-api/v2",
    **kwargs: Any,
) -> None:
    """Initialize authenticated client.

    Args:
        access_key_id: Kalshi access key ID (required)
        private_key_pem: PEM-encoded private key (required)
        base_url: API base URL (defaults to elections API)
        **kwargs: Additional arguments passed to base Client
    """
    self.access_key_id = access_key_id
    self.private_key_pem = private_key_pem
    self.auth = KalshiAuth(self.access_key_id, self.private_key_pem)

    # Initialize base AuthenticatedClient with a dummy token
    # We'll override the authentication in our custom httpx clients
    super().__init__(
        base_url=base_url,
        token="dummy",  # Will be overridden by our custom auth
        **kwargs,
    )

    # Set custom httpx clients with authentication
    self._client = None
    self._async_client = None

Functions

get_httpx_client() -> httpx.Client

Get the underlying authenticated httpx client.

Source code in kalshi_py/auth.py 
163
164
165
166
167
168
169
170
171
172
173
174
175
176
def get_httpx_client(self) -> httpx.Client:
    """Get the underlying authenticated httpx client."""
    if self._client is None:
        self._client = AuthenticatedHTTPXClient(
            kalshi_auth=self.auth,
            base_url=self._base_url,
            cookies=self._cookies,
            headers=self._headers,
            timeout=self._timeout,
            verify=self._verify_ssl,
            follow_redirects=self._follow_redirects,
            **self._httpx_args,
        )
    return self._client

get_async_httpx_client() -> httpx.AsyncClient

Get the underlying authenticated async httpx client.

Source code in kalshi_py/auth.py 
178
179
180
181
182
183
184
185
186
187
188
189
190
191
def get_async_httpx_client(self) -> httpx.AsyncClient:
    """Get the underlying authenticated async httpx client."""
    if self._async_client is None:
        self._async_client = AuthenticatedAsyncHTTPXClient(
            kalshi_auth=self.auth,
            base_url=self._base_url,
            cookies=self._cookies,
            headers=self._headers,
            timeout=self._timeout,
            verify=self._verify_ssl,
            follow_redirects=self._follow_redirects,
            **self._httpx_args,
        )
    return self._async_client

handler: python options: show_source: true show_root_heading: true

Factory Functions

kalshi_py.create_client(base_url: str = 'https://api.elections.kalshi.com/trade-api/v2', access_key_id: str | None = None, private_key_path: str | None = None, private_key_data: str | None = None, **kwargs: object) -> KalshiAuthenticatedClient

Create an authenticated Kalshi client with RSA-PSS signature authentication.

Args:
    base_url: API base URL (defaults to elections API)
    access_key_id: Kalshi access key ID (from environment if not provided)
    private_key_path: Path to private key file (mutually exclusive with private_key_data)
    private_key_data: PEM-encoded private key string (mutually exclusive with private_key_path)
    **kwargs: Additional arguments passed to the client

Returns:
    Authenticated client instance

Example:
    ```python
    from kalshi_py import create_client

    # Using environment variables
    client = create_client()

    # Or with file path
    client = create_client(
        access_key_id="your-key-id",
        private_key_path="/path/to/private-key.pem"
    )

    # Or with PEM data
    client = create_client(
        access_key_id="your-key-id",
        private_key_data="-----BEGIN PRIVATE KEY-----

..." ) ```

Source code in kalshi_py/__init__.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
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
def create_client(
    base_url: str = "https://api.elections.kalshi.com/trade-api/v2",
    access_key_id: str | None = None,
    private_key_path: str | None = None,
    private_key_data: str | None = None,
    **kwargs: object,
) -> KalshiAuthenticatedClient:
    """Create an authenticated Kalshi client with RSA-PSS signature authentication.

    Args:
        base_url: API base URL (defaults to elections API)
        access_key_id: Kalshi access key ID (from environment if not provided)
        private_key_path: Path to private key file (mutually exclusive with private_key_data)
        private_key_data: PEM-encoded private key string (mutually exclusive with private_key_path)
        **kwargs: Additional arguments passed to the client

    Returns:
        Authenticated client instance

    Example:
        ```python
        from kalshi_py import create_client

        # Using environment variables
        client = create_client()

        # Or with file path
        client = create_client(
            access_key_id="your-key-id",
            private_key_path="/path/to/private-key.pem"
        )

        # Or with PEM data
        client = create_client(
            access_key_id="your-key-id",
            private_key_data="-----BEGIN PRIVATE KEY-----\n..."
        )
        ```
    """
    # Get access key ID from parameters or environment
    final_access_key_id = access_key_id or os.getenv("KALSHI_API_KEY_ID")

    if not final_access_key_id:
        raise ValueError("access_key_id must be provided or set in KALSHI_API_KEY_ID environment variable")

    # Handle private key - check for mutual exclusivity
    if private_key_path and private_key_data:
        raise ValueError("Cannot specify both private_key_path and private_key_data")

    if private_key_path:
        # Read from file
        with open(private_key_path) as f:
            final_private_key_data = f.read()
    elif private_key_data:
        # Use provided PEM data
        final_private_key_data = private_key_data
    else:
        # Try environment variable
        final_private_key_data = os.getenv("KALSHI_PY_PRIVATE_KEY_PEM")

    if not final_private_key_data:
        raise ValueError(
            "private_key_path, private_key_data, or KALSHI_PY_PRIVATE_KEY_PEM environment variable must be provided"
        )

    return KalshiAuthenticatedClient(
        base_url=base_url,
        access_key_id=final_access_key_id,
        private_key_pem=final_private_key_data,
        raise_on_unexpected_status=True,
        **kwargs,
    )

handler: python options: show_source: true show_root_heading: true

Error Classes

kalshi_py.errors.UnexpectedStatus(status_code: int, content: bytes)

Bases: Exception

Raised by api functions when the response status an undocumented status and Client.raise_on_unexpected_status is True

Source code in kalshi_py/errors.py 
 7
 8
 9
10
11
12
13
def __init__(self, status_code: int, content: bytes):
    self.status_code = status_code
    self.content = content

    super().__init__(
        f"Unexpected status code: {status_code}\n\nResponse content:\n{content.decode(errors='ignore')}"
    )

handler: python options: show_source: true show_root_heading: true

Usage Examples

Creating a Public Client

from kalshi_py import Client

client = Client(base_url="https://api.elections.kalshi.com/trade-api/v2")

Creating an Authenticated Client

from kalshi_py import create_client

# Using environment variables
client = create_client()

# Or with explicit credentials
client = create_client(
    access_key_id="your-key-id",
    private_key_path="/path/to/private-key.pem"
)

Using the Client

from kalshi_py.api.market import get_markets
response = get_markets.sync(client=client, limit=10)
print(f"Found {len(response.markets)} markets")

Configuration Options

SSL Verification

# Custom SSL certificate
client = create_client(verify_ssl="/path/to/certificate_bundle.pem")

# Disable SSL verification (not recommended for production)
client = create_client(verify_ssl=False)

Timeouts and Headers

import httpx

client = create_client(
    timeout=httpx.Timeout(30.0),
    headers={"User-Agent": "MyApp/1.0"}
)

Request Logging

def log_request(request):
    print(f"Request: {request.method} {request.url}")

def log_response(response):
    print(f"Response: {response.status_code}")

client = create_client(
    httpx_args={
        "event_hooks": {
            "request": [log_request],
            "response": [log_response]
        }
    }
)