Skip to content

Stargate SDK

The Python library for P2P networking in the Traylinx Agent Network. Enable your agents to communicate directly without central coordination.

Current Version: 0.4.0

Installation

pip install traylinx-stargate

pip install "traylinx-stargate[libp2p]"

pip install "traylinx-stargate[all]"

Quick Start

import asyncio
from traylinx_stargate import StarGateNode

async def main():
    # Create and start a node
    node = StarGateNode()
    await node.start()

    print(f"Peer ID: {node.identity.peer_id}")

    # Discover agents with a capability
    peers = await node.discover(capability="text-generation")

    # Call another agent
    if peers:
        response = await node.call(
            peer_id=peers[0].peer_id,
            action="generate",
            payload={"prompt": "Hello!"}
        )
        print(response)

    await node.stop()

asyncio.run(main())

Core Concepts

Identity

Every Stargate node has a unique identity based on an Ed25519 keypair:

from traylinx_stargate import Identity

# Auto-generate or load existing
identity = Identity()
print(identity.peer_id)  # 12D3KooW...

# Generate fresh keypair
identity = Identity(generate=True)

# Load from file
identity = Identity(key_path="~/.traylinx/stargate/identity")

Transport Layers

Stargate supports multiple transport options:

Transport Use Case Pros
NATS Default Reliable, works behind NAT
libp2p Direct P2P No server dependency, Circuit Relay v2

The SDK automatically selects the best transport based on network conditions.

NAT Traversal (Circuit Relay v2)

Agents behind NAT are automatically handled with production-ready features:

  1. NAT Detection: Node detects NAT status on startup (public, private, symmetric NAT)
  2. Direct Connection: Attempts direct peer-to-peer connection first
  3. Circuit Relay Fallback: Uses relay node if direct connection fails
  4. Connection Pooling: Reuses connections efficiently with configurable limits
  5. Health Monitoring: Continuously monitors relay health with automatic failover
  6. Metrics Collection: Comprehensive instrumentation for debugging

Configuration Options: 

node = StarGateNode(
    relay_config={
        "enabled": True,
        "max_connections": 100,
        "health_check_interval": 30,
        "connection_timeout": 10
    }
)

No configuration required for basic usage — it just works with sensible defaults.

Agent Discovery

Find agents by capability:

# Discover by capability
peers = await node.discover(capability="translation")

# Discover by multiple capabilities
peers = await node.discover(capabilities=["translation", "french"])

# Get all online peers
peers = await node.discover()

A2A Communication

Request/Response

response = await node.call(
    peer_id="12D3KooW...",
    action="translate",
    payload={
        "text": "Hello world",
        "target_language": "es"
    }
)

Announce Capabilities

await node.announce(
    capabilities=["text-generation", "summarization"],
    metadata={"model": "gpt-4", "max_tokens": 4096}
)

Listen for Requests

@node.on_request
async def handle_request(request):
    if request.action == "ping":
        return {"pong": True}
    return {"error": "Unknown action"}

Configuration

Environment Variables

Variable Default Description
STARGATE_NATS_URL nats://demo.nats.io:4222 NATS server
STARGATE_TRANSPORT nats Transport layer
STARGATE_IDENTITY_DIR ~/.traylinx/stargate/identity Key storage

Programmatic Configuration

node = StarGateNode(
    nats_url="nats://my-server:4222",
    transport="nats",
    identity_dir="/custom/path"
)

Integration with CLI

The CLI uses Stargate SDK internally:

# CLI command
tx connect

# Equivalent SDK code
node = StarGateNode()
await node.start()

Advanced Features

Relay Node Deployment

Deploy your own relay node for private networks:

# Start a relay node
traylinx stargate relay start \
    --port 4001 \
    --max-connections 1000 \
    --health-check-interval 30

See Relay Node Deployment Guide for details.

Metrics & Monitoring

Access comprehensive metrics:

# Get current metrics
metrics = await node.get_metrics()
print(f"Active connections: {metrics['connections']['active']}")
print(f"Relay usage: {metrics['relay']['messages_relayed']}")
print(f"NAT type: {metrics['nat']['type']}")

Connection Pool Management

Configure connection pooling:

node = StarGateNode(
    pool_config={
        "max_size": 50,
        "min_size": 5,
        "max_idle_time": 300,
        "eviction_interval": 60
    }
)

Next Steps