Async Python client for connecting to a PasarGuard node over gRPC or REST.
This package provides:
- Strongly typed protobuf models (
service_pb2) - Unified node API for both transport types
- User sync helpers (single, batch, and chunked streaming)
- Health/version helpers
- On-demand log streaming
- Node maintenance endpoints (update core/node/geofiles)
pip install pasarguard-node-bridge- Python
>=3.12 - A reachable PasarGuard node
- Node service port (
port) for gRPC or protobuf-REST - Node JSON API port (
api_port) for maintenance endpoints - Server CA certificate content (PEM string)
- API key (UUID string)
import PasarGuardNodeBridge as Bridge
from PasarGuardNodeBridge.common import service_pb2 as servicenode = Bridge.create_node(
connection=Bridge.NodeType.grpc, # Bridge.NodeType.grpc or Bridge.NodeType.rest
address="127.0.0.1",
port=2096, # gRPC or protobuf-REST port (based on connection)
api_port=2097, # REST JSON API port (used internally for maintenance)
server_ca=server_ca_pem_string,
api_key="xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx",
name="node-1", # optional
extra={"region": "eu-1"}, # optional
default_timeout=10, # optional
internal_timeout=15, # optional
proxy="socks5://user:pass@127.0.0.1:1080", # optional
)connection:Bridge.NodeType.grpcorBridge.NodeType.restaddress: node host/IPport: node service portapi_port: node REST JSON API portserver_ca: PEM certificate content as stringapi_key: UUID stringname: optional logger nameextra: optional metadata dictionarylogger: optional custom loggerdefault_timeout: default timeout for public API methodsinternal_timeout: timeout used for internal sync/log operationsproxy: optional upstream proxy URL for node trafficmax_message_size: gRPC only, HTTP/2 window/message sizing
socks5://127.0.0.1:1080socks5://user:pass@127.0.0.1:1080socks4://127.0.0.1:1080http://127.0.0.1:3128http://user:pass@127.0.0.1:3128https://user:pass@proxy.example.com:443
Bridge.NodeType.grpc: gRPC transport viagrpclibBridge.NodeType.rest: protobuf-over-HTTP transport
Use helpers for creating protobuf user/proxy payloads.
user = Bridge.create_user(
email="alice@example.com",
proxies=Bridge.create_proxy(
vmess_id="0d59268a-9847-4218-ae09-65308eb52e08",
vless_id="0d59268a-9847-4218-ae09-65308eb52e08",
vless_flow="",
trojan_password="",
shadowsocks_password="",
shadowsocks_method="",
wireguard_public_key="",
wireguard_peer_ips=["10.10.0.2/32"],
),
inbounds=["inbound-tag-1"],
)For an MTProto node, pass the user's secret (32 hex characters). proxies.mtproto is set only when the secret is not
blank. A limit of None or 0 means no per-user limit.
proxies = Bridge.create_proxy(
mtproto_secret="0123456789abcdef0123456789abcdef",
mtproto_user_ad_tag=None, # optional, 32 hex characters
mtproto_max_tcp_conns=None, # optional
mtproto_max_unique_ips=None, # optional
)You should start() before calling stats/sync/log methods.
await node.start(
config=config_json_string,
backend_type=service.BackendType.XRAY, # or service.BackendType.WIREGUARD / service.BackendType.MTPROTO
users=[user], # optional initial user set
keep_alive=30, # optional
exclude_inbounds=[], # optional
timeout=20,
)
info = await node.info()
print(info.node_version, info.core_version)
await node.stop()update_user and update_users enqueue users and a background worker handles retries and batching.
The worker claims at most sync_batch_size users per delivery (default: 100). Nodes
supporting chunked sync (>=0.2.0) use that transport even for small updates;
older nodes retain the per-user transport. sync_chunk_size controls the number
of users per stream message (default: 100). Both options are positive integers
accepted by create_node(...) and create_node_from_config(..., **runtime_overrides).
sync_lease_seconds (default: 30) bounds the entire queued delivery, including
storage reads, an optional resolver, waiting for the transport lock, and sending
all chunks. Delivery is cancelled before the lease expires. Failed deliveries
and storage errors retry with exponential backoff up to 30 seconds.
A node fails the whole batch when it cannot apply one user in it. Such a rejected
batch is split until the rejected users are alone, so everyone else is still
delivered. A round makes at most 16 sends for this and starts none that could
outlast the round; users not tested by then go back to the queue at once (to be
sent one by one if the round found as many rejected users as accepted ones).
Users the node rejected before are retried as one batch, then one by one if it
is rejected again. The rejected users leave the queue and are queued again after a
backoff, so they never hold back later updates: 2 seconds doubling to 60
seconds; after about 10 minutes an add is retried every 5 minutes (a removal
stays at 60 seconds) and one error is logged for that user. Nothing is dropped.
A retry goes back through the store and is claimed with a lease like any other
update, so a newer update for the same user always wins: the controller calls
the store's request_refresh(node_id, emails) when it has one, otherwise
enqueue_users_if_absent(node_id, users) (the default store has it), otherwise
enqueue_users with the held payload, which could replace a newer update
written by another producer. A newer update through this controller forgets the
user's backoff, and a successful start(...), sync_users(...) or
flush_pending_users() clears all of them. The backoff state lives in this
controller's memory, one entry per user; with several workers on one store each
keeps its own, and a clear on one does not reach the others (their next retry
re-queues the user, which is harmless when the store re-derives current state or
a user_sync_resolver is set). With several controllers on one store, set a
user_sync_resolver (or use a store with request_refresh) so every delivery
reads the current state: without one, an older held payload can still be
delivered after a newer one that the node also rejected and that is waiting in
another controller's backoff. sync_users_chunked sets rejected=True on its
returned failed list for such node rejections; transport errors are retried as
a whole.
disconnect() stops this controller's worker and preserves shared queued work.
connect() wakes the worker to recover it. An external producer that writes
directly to the store should also notify a healthy controller with
await node.wake_sync_worker(); an idle controller is not a permanent queue poller.
await node.update_user(user)
more_users = [user1, user2, user3]
await node.update_users(more_users)By default, queued user updates are kept in a process-local in-memory store shared by node instances. This coordinates controllers in a single worker process when they use the same node_id (or the same service URL when node_id is omitted). For multi-process or multi-host deployments, pass a shared user_sync_store implementation so all workers claim from the same pending-user queue. The package only defines the async protocol; Redis, NATS KV, SQL, or any other backend can be implemented by your application.
store = MyRedisUserSyncStore(redis_client) # implements Bridge.UserSyncStoreProtocol
node = Bridge.create_node(
connection=Bridge.NodeType.grpc,
address="127.0.0.1",
port=2096,
api_port=2097,
server_ca=server_ca_pem_string,
api_key="xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx",
node_id="node-1",
worker_id="worker-a",
user_sync_store=store,
)A UserSyncStoreProtocol implementation must provide these async methods:
enqueue_users(node_id, users)stores latest user payloads by email.claim_users(node_id, worker_id, limit, lease_seconds)atomically leases work and returnsClaimedUseritems.ack_users(node_id, tokens)removes successfully synced claims.requeue_users(node_id, claimed_users)makes failed claims available again.clear(node_id)clears pending and claimed updates for a node.
Delivery is at-least-once. A crashed worker may cause the same latest user payload to be synced again after its lease expires, so external adapters should use atomic claim/lease operations such as Redis Lua/transactions or NATS KV revision compare-and-set.
Claims must exclude concurrent delivery of the same email until release or lease
expiry. Acknowledging an older revision must preserve a newer queued update;
requeuing an expired, cleared or already acknowledged token must not recreate
its old payload. The default store implements these rules and copies protobuf
messages so caller mutations cannot change queued work. An optional
has_pending(node_id) -> bool should include leased work, allowing the worker to
keep polling until it can recover an abandoned lease. Without this optional
method, the worker polls through a lease recovery window after reconnect or a
storage failure.
Adapters that track uncertain acknowledgements may return the affected emails
from ack_users; such adapters must also provide request_refresh(node_id, emails) to durably request delivery of current application state. An optional
refresh_claimed(node_id, claimed_users) can similarly replace work delivered
across a full-sync fence with refresh markers. Without that hook, the worker
requeues those claims using the store's normal revision checks.
Cancellation limits how long this client sends a leased batch. It cannot undo a request that a remote server has already accepted, or provide exactly-once delivery. Process-local storage does not survive process termination; use a durable shared store when that is required.
Applications can pass user_sync_resolver=resolve_users when constructing a
node. This optional async callback receives (node_id, emails) and returns one
current protobuf User for each requested email immediately before delivery.
The library does not access an application database. Return an explicit removal
payload, as appropriate for the node backend, for a deleted user. A missing email
or callback failure retries the batch without sending stale fallback data.
Use a deferred loader when a full snapshot must coexist with queued updates:
async def load_users():
return await application.load_current_node_users(node.node_id)
await node.sync_users_from_source(load_users, timeout=30)This acquires a shared full-sync fence, waits for already claimed deliveries to
finish, captures queued revisions, then calls the loader and sends the snapshot.
Only a successful snapshot retires captured revisions. Updates queued during
the snapshot remain available for subsequent delta delivery. Failures retain
queued work, and loss of the renewable fence aborts the operation. A competing
full sync receives NodeAPIError(409).
The default in-memory store supports this operation across controllers in one
process. External stores must implement SnapshotUserSyncStoreProtocol with
atomic begin_full_sync, renew_full_sync, end_full_sync, fence_state,
capture_queued and retire_captured methods. A fence has an owner token, an
expiry and a monotonically increasing generation. Claims must be blocked while
it is active. Captures return pending revisions and the number of live claims;
retirement removes only matching revisions. Stale owners cannot renew or release
a replacement fence. All controllers sharing a store must honor this contract.
Stores implementing only UserSyncStoreProtocol retain the queue API, but the
new snapshot helper rejects them instead of silently using a local lock.
For custom orchestration, full_sync_fence(), capture_queued_work() and
retire_queued_work(captured) expose the same steps. Keep capture, the authoritative
read, transmission and retirement inside the fence; call the yielded hold's
check() before transmission and retirement. Prefer sync_users_from_source()
unless custom orchestration is necessary.
Existing direct sync_users(...) and sync_users_chunked(...) retain their
behavior. Their explicit flush_pending=True option clears queued work before
sending; use sync_users_from_source() when pending updates must survive a failed
full sync. flush_pending_users() remains an explicit queue discard operation.
Lifecycle operations are coordinated through the same model. The default process-local coordinator prevents concurrent start(), stop(), update_node(), update_core(), and update_geofiles() calls from controllers for the same node in one process. Pass a shared lifecycle_coordinator in multi-process or multi-host deployments so only one worker can perform a lifecycle operation at a time. Read-only status cron jobs can call stats/info normally; if they write shared observed status, use the current lifecycle epoch so stale cron results cannot overwrite a newer reconnect result.
While an operation runs, the controller renews its lease by calling the coordinator's heartbeat(lease) every third of lifecycle_lease_seconds. Return True when the lease was renewed and False when this lease is no longer held (released, or expired and taken by another worker); the controller then stops renewing. An exception from heartbeat is logged and the next renewal is tried as usual.
lifecycle = MyRedisLifecycleCoordinator(redis_client)
node = Bridge.create_node(
connection=Bridge.NodeType.grpc,
address="127.0.0.1",
port=2096,
api_port=2097,
server_ca=server_ca_pem_string,
api_key="xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx",
node_id="node-1",
worker_id="worker-a",
user_sync_store=store,
lifecycle_coordinator=lifecycle,
)
state = await node.get_lifecycle_state()
health = await node.get_health()
if state is not None:
await node.update_observed_lifecycle(
Bridge.LifecycleStatus.HEALTHY if health is Bridge.Health.HEALTHY else Bridge.LifecycleStatus.BROKEN,
expected_epoch=state.epoch,
)A lifecycle adapter must atomically acquire/release leases and fence writes with the returned epoch. This prevents a cron status job or another worker from overwriting the result of a newer start(), stop(), or reconnect flow.
Node connection configs can also be stored through a registry protocol:
registry = MyNodeRegistry(...)
config = Bridge.NodeConfig(
connection="grpc",
address="127.0.0.1",
port=2096,
api_port=2097,
server_ca=server_ca_pem_string,
api_key="xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx",
)
await Bridge.save_node_config(registry, "node-1", config)
node = await Bridge.create_node_from_registry(
registry,
"node-1",
user_sync_store=store,
worker_id="worker-a",
)Use direct sync when you want explicit control in your flow.
await node.sync_users([user1, user2], timeout=15)failed_users = await node.sync_users_chunked(
users=large_user_list,
chunk_size=500,
timeout=30,
)
if failed_users:
print(f"Failed users: {len(failed_users)}")system_stats = await node.get_system_stats()
backend_stats = await node.get_backend_stats()
latencies = await node.get_outbounds_latency()
all_outbounds = await node.get_stats(
stat_type=service.StatType.Outbounds,
reset=False,
)
single_user_online = await node.get_user_online_stats("alice@example.com")
single_user_ips = await node.get_user_online_ip_list("alice@example.com")health = await node.get_health() # Bridge.Health enum
node_ver = await node.node_version()
core_ver = await node.core_version()
node_ver2, core_ver2 = await node.get_versions()
meta = await node.get_extra()stream_logs() yields an asyncio.Queue that contains log lines (str) or Bridge.NodeAPIError.
import asyncio
async with node.stream_logs(max_queue_size=200) as log_queue:
for _ in range(20):
item = await asyncio.wait_for(log_queue.get(), timeout=2)
if isinstance(item, Bridge.NodeAPIError):
raise item
print(item)These methods use the node REST JSON API (api_port).
await node.update_node()
await node.update_core({"version": "latest"})
await node.update_geofiles({"remove_temp": True})Routing operations work over both gRPC and REST. They are xray-only: on a non-xray
(e.g. WireGuard) node the call fails with Bridge.NodeAPIError code 501.
rules = await node.list_routing_rules()
balancer = await node.get_balancer_info("balancer-tag")
route = await node.test_route(
inbound_tag="inbound-1",
network="tcp",
target_domain="example.com",
target_port=443,
)
# `rule` is one xray routing rule as JSON (same shape as a routing.rules[] entry).
# Appended by default (keeps existing rules); pass should_reset=True to clear all
# rules + balancers before adding.
await node.add_routing_rule(
'{"type":"field","outboundTag":"direct","domain":["example.com"],"ruleTag":"r1"}'
)
await node.remove_routing_rule("r1")
await node.override_balancer_target("balancer-tag", "outbound-tag")start(config, backend_type, users, keep_alive=0, exclude_inbounds=[], timeout=None)stop(timeout=None)info(timeout=None)
get_health()node_version()core_version()get_versions()get_extra()
get_system_stats(timeout=None)get_backend_stats(timeout=None)get_stats(stat_type, reset=True, name="", timeout=None)get_outbounds_latency(name="", timeout=None)get_user_online_stats(email, timeout=None)get_user_online_ip_list(email, timeout=None)
update_user(user)(queued/background)update_users(users)(queued/background)sync_users(users, flush_pending=False, timeout=None)(direct)sync_users_chunked(users, chunk_size=100, flush_pending=False, timeout=None)(direct streaming)
Xray-only (gRPC and REST); on a non-xray backend these raise NodeAPIError(501).
list_routing_rules(timeout=None)get_balancer_info(tag, timeout=None)test_route(inbound_tag="", network="", target_ip="", target_domain="", target_port=0, protocol="", user="", attributes=None, field_selectors=None, publish_result=False, timeout=None)add_routing_rule(rule, should_reset=False, timeout=None)remove_routing_rule(rule_tag, timeout=None)override_balancer_target(balancer_tag, target, timeout=None)
stream_logs(max_queue_size=1000)async context manager returning anasyncio.Queue
update_node()update_core(json)update_geofiles(json)
All transport and API errors are surfaced as Bridge.NodeAPIError:
try:
await node.get_backend_stats(timeout=5)
except Bridge.NodeAPIError as e:
print(e.code, e.detail)For direct protobuf usage:
from PasarGuardNodeBridge.common import service_pb2 as serviceimport asyncio
import PasarGuardNodeBridge as Bridge
from PasarGuardNodeBridge.common import service_pb2 as service
async def main():
with open("certs/ssl_cert.pem", "r", encoding="utf-8") as f:
server_ca = f.read()
with open("config/xray.json", "r", encoding="utf-8") as f:
config = f.read()
node = Bridge.create_node(
connection=Bridge.NodeType.grpc,
address="127.0.0.1",
port=2096,
api_port=2097,
server_ca=server_ca,
api_key="d04d8680-942d-4365-992f-9f482275691d",
name="example-node",
)
await node.start(config=config, backend_type=service.BackendType.XRAY, users=[])
print(await node.get_system_stats())
await node.stop()
asyncio.run(main())