Route your broker traffic through a
dedicated IPv6.

# pip install PySocks
import socks, socket

socks.set_default_proxy(
    socks.SOCKS5,
    "comm.servloci.in", 1080,
    username="USER",
    password="PASS",
)
socket.socket = socks.socksocket  # global patch — before any SDK import

# Now every socket in this process egresses via your ServLoci IPv6.
from dhanhq import dhanhq
dhan = dhanhq(client_id="CLIENT_ID", access_token="ACCESS_TOKEN")

Global monkey-patch: applies to any HTTP library that uses socket under the hood — requests, urllib, aiohttp, broker SDKs. Import the SDK after the patch.

import requests

_PROXY = "socks5h://USER:PASS@comm.servloci.in:1080"
_proxies = {"http": _PROXY, "https": _PROXY}

_Base = requests.Session
class _S(_Base):
    def __init__(self, *a, **kw):
        super().__init__(*a, **kw)
        self.proxies.update(_proxies)
requests.Session = _S

# Import the SDK AFTER the patch above.
from kiteconnect import KiteConnect
kite = KiteConnect(api_key="KITE_API_KEY")
kite.set_access_token("ACCESS_TOKEN")

Requires requests[socks]. Scoped to requests.Session — cleaner than the global patch when your process also opens non-broker sockets.

// npm install socks-proxy-agent axios
import axios from 'axios'
import { SocksProxyAgent } from 'socks-proxy-agent'

const agent = new SocksProxyAgent('socks5h://USER:PASS@comm.servloci.in:1080')

const http = axios.create({
  baseURL: 'https://api.kite.trade',
  httpsAgent: agent, httpAgent: agent,
  headers: { Authorization: 'token API_KEY:ACCESS_TOKEN' },
})

const { data } = await http.get('/user/profile')
console.log(data)

curl -s \
  --socks5-hostname USER:PASS@comm.servloci.in:1080 \
  -H "Authorization: token API_KEY:ACCESS_TOKEN" \
  https://api.kite.trade/user/profile

--socks5-hostname resolves DNS on the proxy side (needed for IPv6-only origins).

# pip install servloci requests[socks]
import os
import time
import uuid
from decimal import Decimal

import requests
from servloci import ServLoci

BROKER_ORDER_URL = "https://api.example-broker.com/v1/orders"
MAX_QTY = 10
ALLOWED_SYMBOLS = {"INFY", "TCS", "RELIANCE"}

sl = ServLoci(
    api_key=os.environ["SERVLOCI_API_KEY"],
    api_secret=os.environ["SERVLOCI_API_SECRET"],
)
session = sl.session()
session.headers.update({
    "Authorization": f"Bearer {os.environ['BROKER_ACCESS_TOKEN']}",
    "X-Client-Order-Source": "servloci-course",
})

def decide_signal(snapshot):
    """Replace this with your strategy. Return None to skip trading."""
    if Decimal(snapshot["last_price"]) > Decimal(snapshot["moving_average"]):
        return {"symbol": snapshot["symbol"], "side": "BUY", "qty": 1}
    return None

def risk_check(order):
    if order["symbol"] not in ALLOWED_SYMBOLS:
        raise ValueError("symbol_not_allowed")
    if not 1 <= int(order["qty"]) <= MAX_QTY:
        raise ValueError("qty_outside_limit")
    if order["side"] not in {"BUY", "SELL"}:
        raise ValueError("bad_side")

def place_order(order):
    risk_check(order)
    payload = {
        "symbol": order["symbol"],
        "transaction_type": order["side"],
        "quantity": order["qty"],
        "order_type": "MARKET",
        "product": "INTRADAY",
        "client_order_id": str(uuid.uuid4()),
    }
    response = session.post(BROKER_ORDER_URL, json=payload, timeout=8)
    response.raise_for_status()
    return response.json()

while True:
    snapshot = {"symbol": "INFY", "last_price": "1510.50", "moving_average": "1509.00"}
    signal = decide_signal(snapshot)
    if signal:
        result = place_order(signal)
        print({"signal": signal, "broker_response": result})
    time.sleep(30)