HTTP Gateway API Reference¶
The HTTP Gateway provides a RESTful API for external LLM clients, web applications, and mobile apps to interact with the Adaptive Sentience mesh network. It handles tool execution, mesh discovery, trust management, and observations.
Base URL¶
The default port is 8787, but can be configured via command-line arguments.
Authentication¶
Development Mode¶
For testing and development, you can enable dev-token mode:
This allows local tool execution without capability tokens.
Production Mode¶
Production deployments require capability tokens for authorization:
- Capability tokens must be included in the
tokenfield of tool call requests - Tokens are scoped to specific tools and have time-bound validity
- Ed25519 signatures verify request authenticity
Common Headers¶
All requests should include:
Error Responses¶
Standard error response format:
HTTP status codes:
- 400 - Bad Request (invalid parameters)
- 404 - Not Found
- 500 - Internal Server Error
- 502 - Bad Gateway (upstream node failure)
Endpoints¶
1. Get Tool Catalog¶
Retrieve the catalog of available tools from local and mesh nodes.
Endpoint: GET /v1/catalog
Query Parameters:
| Parameter | Type | Required | Default | Description |
|---|---|---|---|---|
enrich |
boolean | No | false |
Fetch capabilities from each mesh node |
Response:
{
"local": {
"node_id": "node_abc123",
"node_type": "gateway",
"tools": [
{
"name": "pii_redact",
"version": "1.0.0",
"description": "Redact PII from text",
"capability": "tool:pii_redact",
"hash": "sha256:a1b2c3d4e5f6"
}
],
"tools_hash": "sha256:abc123def456"
},
"mesh": [
{
"node_id": "node_xyz789",
"node_type": "edge",
"http_url": "http://192.168.1.100:8000",
"tools_hash": "sha256:def789abc012",
"last_seen": "2026-01-24T12:00:00Z",
"trust_status": "trusted"
}
]
}
cURL Example:
Python Example:
import requests
response = requests.get("http://127.0.0.1:8787/v1/catalog")
catalog = response.json()
print(f"Local tools: {len(catalog['local']['tools'])}")
print(f"Mesh nodes: {len(catalog['mesh'])}")
2. Scan Mesh Network¶
Perform UDP discovery to find active nodes on the local network.
Endpoint: GET /v1/mesh_scan
Query Parameters:
| Parameter | Type | Required | Default | Description |
|---|---|---|---|---|
duration |
float | No | 3.0 |
Scan duration in seconds |
enrich |
boolean | No | false |
Fetch capabilities from discovered nodes |
Response:
{
"nodes": [
{
"node_id": "node_abc123",
"node_type": "edge",
"http_url": "http://192.168.1.100:8000",
"tools_hash": "sha256:abc123",
"public_key": "base64_encoded_key",
"public_key_fingerprint": "sha256:fingerprint",
"location": {
"lat": 37.7749,
"lon": -122.4194,
"accuracy_m": 50
},
"first_seen": "2026-01-24T11:00:00Z",
"last_seen": "2026-01-24T12:00:00Z",
"seen_count": 42
}
],
"count": 1,
"scan_duration": 3.0
}
cURL Example:
Python Example:
import requests
response = requests.get(
"http://127.0.0.1:8787/v1/mesh_scan",
params={"duration": 5.0, "enrich": True}
)
nodes = response.json()["nodes"]
for node in nodes:
print(f"{node['node_id']}: {node['node_type']} at {node['http_url']}")
3. Get Node Capabilities¶
Fetch capabilities and tools from a specific node.
Endpoint: GET /v1/node_capabilities
Query Parameters:
| Parameter | Type | Required | Description |
|---|---|---|---|
url |
string | Yes | Node HTTP URL (e.g., http://192.168.1.100:8000) |
Response:
{
"node_id": "node_abc123",
"node_type": "edge",
"tools": [
{
"name": "unit_convert",
"version": "1.0.0",
"description": "Convert between units",
"required_capability": "tool:unit_convert"
}
],
"location": {
"lat": 37.7749,
"lon": -122.4194,
"accuracy_m": 50
}
}
cURL Example:
Python Example:
import requests
response = requests.get(
"http://127.0.0.1:8787/v1/node_capabilities",
params={"url": "http://192.168.1.100:8000"}
)
capabilities = response.json()
for tool in capabilities["tools"]:
print(f" - {tool['name']} v{tool['version']}")
4. List Agents¶
List trusted nodes with their available tools (for agent selection).
Endpoint: GET /v1/agents
Response:
{
"agents": [
{
"node_id": "node_abc123",
"node_type": "edge",
"http_url": "http://192.168.1.100:8000",
"tools": ["pii_redact", "unit_convert"],
"trust_status": "trusted",
"last_seen": "2026-01-24T12:00:00Z",
"capabilities_ok": true
}
],
"count": 1
}
cURL Example:
Python Example:
import requests
response = requests.get("http://127.0.0.1:8787/v1/agents")
agents = response.json()["agents"]
for agent in agents:
print(f"{agent['node_id']}: {', '.join(agent['tools'])}")
5. Execute Tool Call¶
Execute a tool on local, remote, or auto-routed target.
Endpoint: POST /v1/tool_call
Request Body:
{
"trace_id": "optional_trace_id",
"target": {
"kind": "local",
"http_url": null
},
"tool_name": "pii_redact",
"tool_args": {
"text": "Contact me at john@example.com"
},
"token": null,
"job_id": "optional_job_id_for_idempotency"
}
Target Types:
| Kind | Description | http_url Required |
|---|---|---|
local |
Execute on gateway's local toolhost | No |
remote |
Execute on specific remote node | Yes |
auto |
Auto-route to best available node | No |
Request Fields:
| Field | Type | Required | Description |
|---|---|---|---|
trace_id |
string | No | Trace ID for request tracking |
target |
object | Yes | Execution target specification |
target.kind |
string | Yes | One of: local, remote, auto |
target.http_url |
string | Conditional | Required if kind is remote |
tool_name |
string | Yes | Name of tool to execute |
tool_args |
object | Yes | Tool-specific arguments |
token |
object | No | Capability token (optional in dev mode) |
job_id |
string | No | Job ID for idempotent execution |
Response:
{
"ok": true,
"trace_id": "trace_abc123",
"target": {
"kind": "local"
},
"tool_name": "pii_redact",
"tool_version": "1.0.0",
"verified": false,
"result": {
"redacted_text": "Contact me at [REDACTED]",
"redactions": [
{
"type": "email",
"count": 1
}
]
},
"error": null,
"execution_path": ["local"],
"metadata": {
"node_id": "node_gateway",
"latency_ms": 12
},
"degraded": false,
"final_target": null,
"primary_error": null
}
Response Fields:
| Field | Type | Description |
|---|---|---|
ok |
boolean | Whether execution succeeded |
trace_id |
string | Trace ID for request tracking |
target |
object | Requested target specification |
tool_name |
string | Name of executed tool |
tool_version |
string | Version of executed tool |
verified |
boolean | Whether response signature was verified |
result |
object | Tool execution result (null on error) |
error |
any | Error details (null on success) |
execution_path |
array | List of execution steps |
metadata |
object | Execution metadata (node_id, latency, etc.) |
degraded |
boolean | True if failover occurred |
final_target |
object | Actual target used (if different from requested) |
primary_error |
string | Error from primary target (if failed over) |
cURL Example (Local):
curl -X POST http://127.0.0.1:8787/v1/tool_call \
-H "Content-Type: application/json" \
-d '{
"target": {"kind": "local"},
"tool_name": "pii_redact",
"tool_args": {
"text": "Call me at 555-123-4567"
}
}'
cURL Example (Remote):
curl -X POST http://127.0.0.1:8787/v1/tool_call \
-H "Content-Type: application/json" \
-d '{
"target": {
"kind": "remote",
"http_url": "http://192.168.1.100:8000"
},
"tool_name": "unit_convert",
"tool_args": {
"value": 10,
"from_unit": "miles",
"to_unit": "km"
}
}'
cURL Example (Auto-routing with Failover):
curl -X POST http://127.0.0.1:8787/v1/tool_call \
-H "Content-Type: application/json" \
-d '{
"target": {"kind": "auto"},
"tool_name": "pii_redact",
"tool_args": {
"text": "Email: admin@company.com"
}
}'
Python Example:
import requests
response = requests.post(
"http://127.0.0.1:8787/v1/tool_call",
json={
"target": {"kind": "local"},
"tool_name": "unit_convert",
"tool_args": {
"value": 100,
"from_unit": "celsius",
"to_unit": "fahrenheit"
}
}
)
result = response.json()
if result["ok"]:
print(f"Result: {result['result']['value']} {result['result']['to_unit']}")
print(f"Latency: {result['metadata']['latency_ms']}ms")
else:
print(f"Error: {result['error']}")
Error Handling Example:
import requests
response = requests.post(
"http://127.0.0.1:8787/v1/tool_call",
json={
"target": {"kind": "auto"},
"tool_name": "pii_redact",
"tool_args": {"text": "Sensitive data here"}
}
)
result = response.json()
if result["degraded"]:
print(f"Warning: Failover occurred")
print(f"Primary error: {result['primary_error']}")
print(f"Final target: {result['final_target']}")
if not result["ok"]:
print(f"Execution failed: {result['error']}")
6. Orchestrate Workflow¶
Execute a natural language workflow with automatic planning.
Endpoint: POST /v1/orchestrate
Request Body:
Response:
{
"ok": true,
"steps": [
{
"tool_name": "pii_redact",
"result": {
"redacted_text": "Redact PII and convert 10 miles to km. Email me at [REDACTED]"
}
},
{
"tool_name": "unit_convert",
"result": {
"value": 16.0934,
"from_unit": "miles",
"to_unit": "km"
}
}
],
"execution_proof": {
"workflow_id": "workflow_abc123",
"timeline": [
{
"step_index": 0,
"tool_name": "pii_redact",
"node_id": "node_abc",
"verified": true,
"timestamp": "2026-01-24T12:00:00Z"
}
]
}
}
cURL Example:
curl -X POST http://127.0.0.1:8787/v1/orchestrate \
-H "Content-Type: application/json" \
-d '{
"text": "Redact PII and summarize: Contact support@company.com for help"
}'
Python Example:
import requests
response = requests.post(
"http://127.0.0.1:8787/v1/orchestrate",
json={
"text": "Validate this data and redact PII: Contact admin@example.com"
}
)
result = response.json()
if result["ok"]:
for i, step in enumerate(result["steps"]):
print(f"Step {i+1}: {step['tool_name']}")
print(f" Result: {step['result']}")
7. Get Mesh Snapshot¶
Get a comprehensive snapshot of the mesh network with location data.
Endpoint: GET /v1/mesh_snapshot
Query Parameters:
| Parameter | Type | Required | Default | Description |
|---|---|---|---|---|
duration |
float | No | 3.0 |
Scan duration in seconds |
Response:
{
"nodes": [
{
"node_id": "node_abc123",
"node_type": "edge",
"http_url": "http://192.168.1.100:8000",
"trust_status": "trusted",
"location": {
"lat": 37.7749,
"lon": -122.4194,
"accuracy_m": 50
},
"last_seen": "2026-01-24T12:00:00Z"
}
],
"observations": {
"ble_devices": [
{
"address": "AA:BB:CC:DD:EE:FF",
"name": "Bluetooth Speaker",
"rssi": -65
}
],
"wifi_networks": [
{
"ssid": "CoffeeShop_WiFi",
"bssid": "00:11:22:33:44:55",
"rssi": -45
}
]
}
}
cURL Example:
Python Example:
import requests
response = requests.get(
"http://127.0.0.1:8787/v1/mesh_snapshot",
params={"duration": 5.0}
)
snapshot = response.json()
print(f"Nodes: {len(snapshot['nodes'])}")
print(f"BLE devices: {len(snapshot['observations']['ble_devices'])}")
print(f"WiFi networks: {len(snapshot['observations']['wifi_networks'])}")
8. Get Observations¶
Retrieve stored BLE and WiFi observations.
Endpoint: GET /v1/observations
Query Parameters:
| Parameter | Type | Required | Default | Description |
|---|---|---|---|---|
limit |
integer | No | 100 |
Maximum number of results |
source |
string | No | all |
Filter by source: ble, wifi, or all |
Response:
{
"ble_devices": [
{
"address": "AA:BB:CC:DD:EE:FF",
"name": "Device Name",
"rssi": -65,
"observed_at": "2026-01-24T12:00:00Z",
"node_id": "node_abc123"
}
],
"wifi_networks": [
{
"ssid": "NetworkName",
"bssid": "00:11:22:33:44:55",
"rssi": -45,
"observed_at": "2026-01-24T12:00:00Z",
"node_id": "node_abc123"
}
],
"count": {
"ble": 1,
"wifi": 1
}
}
cURL Example:
Python Example:
import requests
response = requests.get(
"http://127.0.0.1:8787/v1/observations",
params={"limit": 50, "source": "wifi"}
)
observations = response.json()
for wifi in observations["wifi_networks"]:
print(f"{wifi['ssid']}: {wifi['rssi']} dBm")
9. Ingest Observations¶
Submit BLE or WiFi observations from Android collectors.
Endpoint: POST /v1/observations/ingest
Request Body:
{
"source": "ble",
"items": [
{
"address": "AA:BB:CC:DD:EE:FF",
"name": "Bluetooth Speaker",
"rssi": -65
}
],
"context": {
"collector": "android_app",
"location": {
"lat": 37.7749,
"lon": -122.4194
}
}
}
Request Fields:
| Field | Type | Required | Description |
|---|---|---|---|
source |
string | Yes | One of: ble, wifi |
items |
array | Yes | List of observation objects |
context |
object | No | Additional context metadata |
Response:
cURL Example:
curl -X POST http://127.0.0.1:8787/v1/observations/ingest \
-H "Content-Type: application/json" \
-d '{
"source": "wifi",
"items": [
{
"ssid": "HomeNetwork",
"bssid": "00:11:22:33:44:55",
"rssi": -40
}
]
}'
Python Example:
import requests
response = requests.post(
"http://127.0.0.1:8787/v1/observations/ingest",
json={
"source": "ble",
"items": [
{
"address": "AA:BB:CC:DD:EE:FF",
"name": "Fitness Tracker",
"rssi": -70
}
],
"context": {
"collector": "python_script"
}
}
)
print(response.json())
10. Update Node Location¶
Update the gateway's location cache.
Endpoint: POST /v1/node_location/update
Request Body:
Request Fields:
| Field | Type | Required | Default | Description |
|---|---|---|---|---|
lat |
float | Yes | - | Latitude in decimal degrees |
lon |
float | Yes | - | Longitude in decimal degrees |
accuracy_m |
float | No | 100 |
Accuracy in meters |
Response:
cURL Example:
curl -X POST http://127.0.0.1:8787/v1/node_location/update \
-H "Content-Type: application/json" \
-d '{
"lat": 37.7749,
"lon": -122.4194,
"accuracy_m": 10
}'
Python Example:
import requests
response = requests.post(
"http://127.0.0.1:8787/v1/node_location/update",
json={
"lat": 37.7749,
"lon": -122.4194,
"accuracy_m": 25
}
)
print(response.json())
Trust Management Endpoints¶
11. Get Trust Bundle¶
Get the gateway's trust bundle for QR code pairing.
Endpoint: GET /v1/trust/bundle
Query Parameters:
| Parameter | Type | Required | Default | Description |
|---|---|---|---|---|
format |
string | No | json |
Output format: json or qr |
Response (JSON format):
{
"node_id": "node_abc123",
"public_key": "base64_encoded_ed25519_key",
"public_key_fingerprint": "sha256:abc123def456",
"node_type": "gateway",
"label": "My Gateway",
"x25519_pubkey": "base64_encoded_x25519_key",
"http_url": "http://192.168.1.5:8787"
}
Response (QR format):
Returns ASCII art QR code for scanning.
cURL Example:
Python Example:
import requests
# Get JSON bundle
response = requests.get("http://127.0.0.1:8787/v1/trust/bundle")
bundle = response.json()
print(f"Node ID: {bundle['node_id']}")
print(f"Fingerprint: {bundle['public_key_fingerprint']}")
# Get QR code
response = requests.get(
"http://127.0.0.1:8787/v1/trust/bundle",
params={"format": "qr"}
)
print(response.text) # ASCII QR code
12. List Trusted Nodes¶
List all nodes in the trust store.
Endpoint: GET /v1/trust/list
Response:
{
"trusted": [
{
"node_id": "node_abc123",
"public_key_fingerprint": "sha256:abc123",
"label": "Edge Node 1",
"http_url": "http://192.168.1.100:8000",
"added_at": "2026-01-24T10:00:00Z"
}
],
"blocked": [
{
"node_id": "node_xyz789",
"public_key_fingerprint": "sha256:xyz789",
"reason": "Suspicious behavior",
"blocked_at": "2026-01-24T11:00:00Z"
}
],
"count": {
"trusted": 1,
"blocked": 1
}
}
cURL Example:
Python Example:
import requests
response = requests.get("http://127.0.0.1:8787/v1/trust/list")
trust = response.json()
print(f"Trusted nodes: {trust['count']['trusted']}")
for node in trust["trusted"]:
print(f" - {node['label']} ({node['node_id']})")
13. Add Trusted Node¶
Add a node to the trust store.
Endpoint: POST /v1/trust/add
Request Body:
{
"node_id": "node_abc123",
"public_key": "base64_encoded_ed25519_key",
"public_key_fingerprint": "sha256:abc123def456",
"label": "Edge Node 1",
"x25519_pubkey": "base64_encoded_x25519_key",
"http_url": "http://192.168.1.100:8000"
}
Request Fields:
| Field | Type | Required | Description |
|---|---|---|---|
node_id |
string | Yes | Node identifier |
public_key |
string | Yes | Base64-encoded Ed25519 public key |
public_key_fingerprint |
string | Yes | SHA256 fingerprint (sha256:...) |
label |
string | No | User-friendly label |
x25519_pubkey |
string | No | Base64-encoded X25519 public key for E2EE |
http_url |
string | No | HTTP URL for mesh transport |
Response:
cURL Example:
curl -X POST http://127.0.0.1:8787/v1/trust/add \
-H "Content-Type: application/json" \
-d '{
"node_id": "node_edge_1",
"public_key": "AAAA...base64...ZZZZ",
"public_key_fingerprint": "sha256:abcdef123456",
"label": "Edge Node 1",
"http_url": "http://192.168.1.100:8000"
}'
Python Example:
import requests
response = requests.post(
"http://127.0.0.1:8787/v1/trust/add",
json={
"node_id": "node_edge_1",
"public_key": "AAAA...base64...ZZZZ",
"public_key_fingerprint": "sha256:abcdef123456",
"label": "Raspberry Pi Edge Node",
"http_url": "http://192.168.1.100:8000"
}
)
print(response.json())
14. Block Node¶
Block a node in the trust store.
Endpoint: POST /v1/trust/block
Request Body:
{
"node_id": "node_xyz789",
"public_key_fingerprint": "sha256:xyz789",
"reason": "Suspicious behavior detected"
}
Response:
cURL Example:
curl -X POST http://127.0.0.1:8787/v1/trust/block \
-H "Content-Type: application/json" \
-d '{
"node_id": "node_malicious",
"public_key_fingerprint": "sha256:badactor",
"reason": "Failed signature verification"
}'
Python Example:
import requests
response = requests.post(
"http://127.0.0.1:8787/v1/trust/block",
json={
"node_id": "node_suspicious",
"public_key_fingerprint": "sha256:suspicious123",
"reason": "Repeated timeout failures"
}
)
print(response.json())
15. Remove Node from Trust Store¶
Remove a node from trust store (neither trusted nor blocked).
Endpoint: POST /v1/trust/remove
Request Body:
Response:
cURL Example:
curl -X POST http://127.0.0.1:8787/v1/trust/remove \
-H "Content-Type: application/json" \
-d '{"node_id": "node_old_device"}'
Python Example:
import requests
response = requests.post(
"http://127.0.0.1:8787/v1/trust/remove",
json={"node_id": "node_old_device"}
)
print(response.json())
Provisioning Endpoints¶
16. Get Provisioning Package¶
Generate a provisioning package for new nodes.
Endpoint: GET /v1/provision/package
Query Parameters:
| Parameter | Type | Required | Default | Description |
|---|---|---|---|---|
label |
string | No | "New Node" |
Label for the new node |
format |
string | No | json |
Output format: json or qr |
Response (JSON format):
{
"join_token": "unique_join_token_string",
"gateway_node_id": "node_gateway_abc",
"gateway_pubkey": "base64_encoded_key",
"gateway_fingerprint": "sha256:abc123",
"gateway_http_url": "http://192.168.1.5:8787",
"created_at": "2026-01-24T12:00:00Z",
"expires_at": "2026-01-24T13:00:00Z"
}
cURL Example:
Python Example:
import requests
response = requests.get(
"http://127.0.0.1:8787/v1/provision/package",
params={"label": "Android_Device_1", "format": "json"}
)
package = response.json()
print(f"Join token: {package['join_token']}")
print(f"Expires: {package['expires_at']}")
17. Join Mesh via Provisioning¶
New node joins the mesh using a provisioning package.
Endpoint: POST /v1/provision/join
Request Body:
{
"join_token": "unique_join_token_string",
"node_bundle": {
"node_id": "node_new_123",
"public_key": "base64_encoded_key",
"public_key_fingerprint": "sha256:new123",
"x25519_pubkey": "base64_encoded_x25519_key",
"http_url": "http://192.168.1.110:8000",
"label": "New Edge Node"
}
}
Response:
cURL Example:
curl -X POST http://127.0.0.1:8787/v1/provision/join \
-H "Content-Type: application/json" \
-d '{
"join_token": "abc123xyz789",
"node_bundle": {
"node_id": "node_rpi_001",
"public_key": "AAAA...ZZZZ",
"public_key_fingerprint": "sha256:rpi001",
"http_url": "http://192.168.1.110:8000",
"label": "Raspberry Pi 001"
}
}'
Python Example:
import requests
response = requests.post(
"http://127.0.0.1:8787/v1/provision/join",
json={
"join_token": "provisioning_token_here",
"node_bundle": {
"node_id": "node_android_1",
"public_key": "base64_pubkey",
"public_key_fingerprint": "sha256:android1",
"http_url": "http://192.168.1.120:8000",
"label": "Android Phone"
}
}
)
print(response.json())
18. List Pending Provisioning Tokens¶
List all active provisioning tokens.
Endpoint: GET /v1/provision/pending
Response:
{
"tokens": [
{
"join_token": "abc123xyz789",
"label": "New Edge Node",
"created_at": "2026-01-24T12:00:00Z",
"expires_at": "2026-01-24T13:00:00Z",
"used": false
}
],
"count": 1
}
cURL Example:
Python Example:
import requests
response = requests.get("http://127.0.0.1:8787/v1/provision/pending")
tokens = response.json()["tokens"]
for token in tokens:
print(f"{token['label']}: {token['join_token']} (expires: {token['expires_at']})")
19. Health Check¶
Check gateway health and status.
Endpoint: GET /health
Response:
cURL Example:
Python Example:
import requests
response = requests.get("http://127.0.0.1:8787/health")
health = response.json()
if health["status"] == "ok":
print(f"Gateway is healthy (uptime: {health['uptime']}s)")
else:
print("Gateway is unhealthy!")
Complete Python Client Example¶
Here's a complete example showing common gateway operations:
import requests
import json
from typing import Dict, Any, List
class AdaptiveSentienceClient:
"""Python client for Adaptive Sentience HTTP Gateway."""
def __init__(self, base_url: str = "http://127.0.0.1:8787"):
self.base_url = base_url.rstrip("/")
self.session = requests.Session()
def get_catalog(self, enrich: bool = False) -> Dict[str, Any]:
"""Get tool catalog."""
response = self.session.get(
f"{self.base_url}/v1/catalog",
params={"enrich": enrich}
)
response.raise_for_status()
return response.json()
def scan_mesh(self, duration: float = 3.0, enrich: bool = False) -> List[Dict]:
"""Scan mesh network for nodes."""
response = self.session.get(
f"{self.base_url}/v1/mesh_scan",
params={"duration": duration, "enrich": enrich}
)
response.raise_for_status()
return response.json()["nodes"]
def execute_tool(
self,
tool_name: str,
tool_args: Dict[str, Any],
target_kind: str = "local",
http_url: str = None
) -> Dict[str, Any]:
"""Execute a tool."""
request = {
"target": {"kind": target_kind},
"tool_name": tool_name,
"tool_args": tool_args
}
if http_url:
request["target"]["http_url"] = http_url
response = self.session.post(
f"{self.base_url}/v1/tool_call",
json=request
)
response.raise_for_status()
return response.json()
def orchestrate(self, text: str) -> Dict[str, Any]:
"""Orchestrate a natural language workflow."""
response = self.session.post(
f"{self.base_url}/v1/orchestrate",
json={"text": text}
)
response.raise_for_status()
return response.json()
def list_agents(self) -> List[Dict]:
"""List trusted agents with their tools."""
response = self.session.get(f"{self.base_url}/v1/agents")
response.raise_for_status()
return response.json()["agents"]
def add_trusted_node(
self,
node_id: str,
public_key: str,
fingerprint: str,
label: str = None,
http_url: str = None
) -> Dict[str, Any]:
"""Add a node to trust store."""
request = {
"node_id": node_id,
"public_key": public_key,
"public_key_fingerprint": fingerprint
}
if label:
request["label"] = label
if http_url:
request["http_url"] = http_url
response = self.session.post(
f"{self.base_url}/v1/trust/add",
json=request
)
response.raise_for_status()
return response.json()
def get_trust_bundle(self, format: str = "json") -> Any:
"""Get gateway trust bundle."""
response = self.session.get(
f"{self.base_url}/v1/trust/bundle",
params={"format": format}
)
response.raise_for_status()
if format == "qr":
return response.text
return response.json()
# Example usage
if __name__ == "__main__":
client = AdaptiveSentienceClient()
# Get catalog
catalog = client.get_catalog()
print(f"Local tools: {len(catalog['local']['tools'])}")
# Execute tool
result = client.execute_tool(
tool_name="unit_convert",
tool_args={
"value": 100,
"from_unit": "celsius",
"to_unit": "fahrenheit"
}
)
if result["ok"]:
print(f"100°C = {result['result']['value']}°F")
# Orchestrate workflow
workflow = client.orchestrate(
"Redact PII and convert 10 miles to km. Contact: admin@example.com"
)
print(f"Workflow steps: {len(workflow['steps'])}")
# List agents
agents = client.list_agents()
for agent in agents:
print(f"{agent['node_id']}: {', '.join(agent['tools'])}")
Rate Limiting and Performance¶
The gateway does not currently implement rate limiting, but production deployments should consider:
- Connection pooling: Reuse HTTP connections for multiple requests
- Batch operations: Group multiple tool calls when possible
- Caching: Cache catalog and mesh scan results (TTL: 30s for capabilities, 5s for mesh)
- Timeouts: Default timeout is 10s for remote tool calls
Security Best Practices¶
- Use HTTPS in production: Deploy behind a reverse proxy with TLS
- Enable capability tokens: Never use
--dev-tokenin production - Verify signatures: Check
verified: truein tool call responses - Trust store management: Regularly audit trusted nodes
- Network isolation: Run gateway on isolated network segment
- Audit logging: Monitor tool execution and trust changes
Error Codes¶
| HTTP Status | Meaning | Common Causes |
|---|---|---|
| 400 | Bad Request | Invalid parameters, missing required fields |
| 404 | Not Found | Endpoint doesn't exist |
| 500 | Internal Server Error | Gateway malfunction, database error |
| 502 | Bad Gateway | Remote node unreachable or timeout |
| 503 | Service Unavailable | Gateway overloaded or shutting down |
Changelog¶
- v1.0.0 (2026-01-24): Initial API release
- Tool catalog and mesh discovery
- Tool execution with auto-routing
- Trust management and provisioning
- Observations ingestion
- Natural language orchestration