SnipeRouteClient Reference

Overview

SnipeRouteClient is the main class for interacting with the SnipeRoute API. It provides methods for creating and querying Trade Intents.

Initialization

Basic

from sniperoute import SnipeRouteClient
 
# Automatically loads from environment variables
client = SnipeRouteClient()

With Configuration

client = SnipeRouteClient(
    base_url="https://api.sniperoute.io",
    api_key="sk_live_your_api_key_here",
    timeout=30.0,
    max_retries=3
)

Constructor Parameters

Parameter	Type	Default	Description
`base_url`	str \| None	`SNIPEROUTE_API_URL` env var	API base URL
`api_key`	str \| None	`SNIPEROUTE_API_KEY` env var	API key
`timeout`	float	30.0	Request timeout in seconds
`max_retries`	int	3	Max retry attempts for transient failures

Methods

create_intent()

Create a new Trade Intent.

async def create_intent(
    intent: TradeIntentRequest
) -> TradeIntentResponse

Parameters:

intent (TradeIntentRequest): Trade Intent to create

Returns:

TradeIntentResponse: Created Trade Intent

Raises:

ValidationError: Invalid parameters
ConflictError: Duplicate intent_id
AuthenticationError: Invalid API key
SnipeRouteAPIError: Other API errors

Example:

from sniperoute.models import TradeIntentRequest, OrderSide, OrderType
from decimal import Decimal
 
intent = TradeIntentRequest(
    intent_id="trade_001",
    symbol="AAPL",
    side=OrderSide.BUY,
    quantity=Decimal("10"),
    order_type=OrderType.MARKET,
    broker_id="my_broker"
)
 
response = await client.create_intent(intent)
print(f"Created: {response.intent_id}")

get_intent_by_id()

Get Trade Intent by internal UUID.

async def get_intent_by_id(
    intent_id: str
) -> TradeIntentResponse

Parameters:

intent_id (str): Internal UUID

Returns:

TradeIntentResponse: Trade Intent

Raises:

NotFoundError: Intent not found
AuthenticationError: Invalid API key
SnipeRouteAPIError: Other API errors

Example:

intent = await client.get_intent_by_id(
    "550e8400-e29b-41d4-a716-446655440000"
)
print(f"Status: {intent.status}")

get_intent_by_external_id()

Get Trade Intent by your external intent_id.

async def get_intent_by_external_id(
    external_id: str
) -> TradeIntentResponse

Parameters:

external_id (str): Your external intent identifier

Returns:

TradeIntentResponse: Trade Intent

Raises:

NotFoundError: Intent not found
AuthenticationError: Invalid API key
SnipeRouteAPIError: Other API errors

Example:

# Recommended: Use external ID
intent = await client.get_intent_by_external_id("trade_001")
print(f"Status: {intent.status}")

Best Practice: Use get_intent_by_external_id() instead of get_intent_by_id() when possible.

Webhook Management Methods

create_webhook()

Create a new webhook.

async def create_webhook(
    webhook: WebhookCreate
) -> WebhookCreateResponse

Parameters:

webhook (WebhookCreate): Webhook configuration (url, events, description)

Returns:

WebhookCreateResponse: Created webhook with signing secret (shown only once!)

Example:

from sniperoute import WebhookCreate
 
webhook = WebhookCreate(
    url="https://api.yourapp.com/webhooks/sniperoute",
    events=["intent.filled", "intent.rejected"],
    description="Production webhook"
)
 
response = await client.create_webhook(webhook)
print(f"Webhook ID: {response.id}")
print(f"Secret: {response.secret}")  # Save this! Only shown once

list_webhooks()

List all registered webhooks.

async def list_webhooks() -> list[WebhookResponse]

Returns:

List of WebhookResponse: Webhook configurations

Example:

webhooks = await client.list_webhooks()
for webhook in webhooks:
    print(f"{webhook.id}: {webhook.url} ({', '.join(webhook.events)})")

get_webhook()

Get a webhook by ID.

async def get_webhook(
    webhook_id: str
) -> WebhookResponse

Parameters:

webhook_id (str): Webhook ID

Returns:

WebhookResponse: Webhook configuration

update_webhook()

Update a webhook.

async def update_webhook(
    webhook_id: str,
    update: WebhookUpdate
) -> WebhookResponse

Parameters:

webhook_id (str): Webhook ID
update (WebhookUpdate): Fields to update (url, events, description, is_active)

Returns:

WebhookResponse: Updated webhook configuration

Example:

from sniperoute import WebhookUpdate
 
# Disable webhook
update = WebhookUpdate(is_active=False)
updated = await client.update_webhook("wh_abc123", update)
 
# Change events
update = WebhookUpdate(events=["intent.filled", "intent.partially_filled"])
updated = await client.update_webhook("wh_abc123", update)

delete_webhook()

Delete a webhook.

async def delete_webhook(webhook_id: str) -> None

Parameters:

webhook_id (str): Webhook ID

rotate_webhook_secret()

Rotate webhook signing secret. The old secret immediately becomes invalid.

async def rotate_webhook_secret(
    webhook_id: str
) -> WebhookSecretResponse

Parameters:

webhook_id (str): Webhook ID

Returns:

WebhookSecretResponse: New signing secret

Example:

response = await client.rotate_webhook_secret("wh_abc123")
print(f"New secret: {response.secret}")  # Update your webhook handler!

list_webhook_deliveries()

Get delivery history for a webhook.

async def list_webhook_deliveries(
    webhook_id: str,
    limit: int = 50,
    offset: int = 0
) -> list[WebhookDelivery]

Parameters:

webhook_id (str): Webhook ID
limit (int): Maximum number of deliveries to return (default: 50)
offset (int): Number of deliveries to skip (default: 0)

Returns:

List of WebhookDelivery: Delivery attempts with status, response, etc.

Example:

deliveries = await client.list_webhook_deliveries("wh_abc123", limit=10)
for delivery in deliveries:
    print(f"{delivery.event_type}: {delivery.status} ({delivery.http_status})")

send_test_webhook()

Send a test event to a webhook.

async def send_test_webhook(
    webhook_id: str,
    event_type: str = "test.ping"
) -> WebhookTestResponse

Parameters:

webhook_id (str): Webhook ID
event_type (str): Event type to send (default: "test.ping")

Returns:

WebhookTestResponse: Test delivery information

Example:

response = await client.send_test_webhook("wh_abc123")
print(f"Test event sent: {response.event_id}")

Configuration

Environment Variables

The client automatically loads configuration from these environment variables:

SNIPEROUTE_API_KEY=sk_live_your_api_key_here
SNIPEROUTE_API_URL=https://api.sniperoute.io

Custom Timeout

# Longer timeout for slow connections
client = SnipeRouteClient(timeout=60.0)

Disable Retries

# No automatic retries
client = SnipeRouteClient(max_retries=0)

Context Manager

Use async with for automatic cleanup:

async with SnipeRouteClient() as client:
    intent = TradeIntentRequest(...)
    response = await client.create_intent(intent)
    print(f"Created: {response.intent_id}")
 
# Client automatically closed

Error Handling

Try/Except

from sniperoute.exceptions import (
    ValidationError,
    ConflictError,
    NotFoundError,
    AuthenticationError,
    RateLimitError,
    SnipeRouteAPIError
)
 
try:
    response = await client.create_intent(intent)
except ValidationError as e:
    print(f"Validation error: {e.message}")
except ConflictError as e:
    print(f"Duplicate intent_id: {e.message}")
except NotFoundError as e:
    print(f"Not found: {e.message}")
except AuthenticationError as e:
    print(f"Auth error: {e.message}")
except RateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after}s")
except SnipeRouteAPIError as e:
    print(f"API error: {e.message}")

Advanced Usage

Custom Headers

from sniperoute import SnipeRouteClient
 
class CustomClient(SnipeRouteClient):
    def _get_headers(self) -> dict[str, str]:
        headers = super()._get_headers()
        headers["X-Custom-Header"] = "value"
        return headers
 
client = CustomClient()

Retry Configuration

import asyncio
from sniperoute import SnipeRouteClient
from sniperoute.exceptions import SnipeRouteAPIError
 
async def create_with_retry(client, intent, max_attempts=3):
    for attempt in range(max_attempts):
        try:
            return await client.create_intent(intent)
        except SnipeRouteAPIError as e:
            if e.status_code >= 500 and attempt < max_attempts - 1:
                wait_time = 2 ** attempt
                await asyncio.sleep(wait_time)
            else:
                raise

Logging

import logging
from sniperoute import SnipeRouteClient
 
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger("sniperoute")
 
client = SnipeRouteClient()
 
# SDK will log HTTP requests/responses

Complete Example

import asyncio
import logging
from sniperoute import SnipeRouteClient
from sniperoute.models import TradeIntentRequest, OrderSide, OrderType
from sniperoute.exceptions import (
    ValidationError,
    ConflictError,
    RateLimitError,
    SnipeRouteAPIError
)
from decimal import Decimal
 
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
 
async def main():
    # Initialize client with custom config
    async with SnipeRouteClient(
        timeout=60.0,
        max_retries=3
    ) as client:
        try:
            # Create Trade Intent
            intent = TradeIntentRequest(
                intent_id="advanced_example_001",
                symbol="NVDA",
                side=OrderSide.BUY,
                quantity=Decimal("20"),
                order_type=OrderType.LIMIT,
                limit_price=Decimal("500.00"),
                broker_id="my_broker"
            )
 
            logger.info(f"Submitting intent: {intent.intent_id}")
            response = await client.create_intent(intent)
 
            logger.info(f"Intent created: {response.intent_id}")
            logger.info(f"Status: {response.status}")
 
            # Check status
            status = await client.get_intent_by_external_id(
                "advanced_example_001"
            )
 
            if status.status == "filled":
                logger.info("Order filled!")
                for order in status.orders:
                    for fill in order.fills:
                        logger.info(f"   {fill.quantity} @ ${fill.price}")
            else:
                logger.info(f"Status: {status.status}")
 
        except ValidationError as e:
            logger.error(f"Validation error: {e.message}")
        except ConflictError as e:
            logger.error(f"Duplicate intent_id: {e.message}")
        except RateLimitError as e:
            logger.warning(f"Rate limited. Retry after {e.retry_after}s")
        except SnipeRouteAPIError as e:
            logger.error(f"API error: {e.message}")
            logger.error(f"Request ID: {e.request_id}")
 
if __name__ == "__main__":
    asyncio.run(main())

Next Steps

Webhooks

Receive and verify webhook events

Models

Pydantic models reference

Error Handling

Exception hierarchy and handling

Quickstart Webhooks