nexusos-sdk 1.0.1

Official Python SDK for NexusOS AI Agent Governance

NexusOS Python SDK

Official Python SDK for NexusOS AI Agent Governance. Monitor, log, and govern AI agents with ease.

Installation

pip install nexusos-sdk

Quick Start

from nexusos import NexusAgent

agent = NexusAgent(
    api_key='nxs_live_xxxx',
    agent_id='nexus_agt_xxxx'
)

# Log an LLM call
agent.log_llm_call(
    model='gpt-4',
    tokens_used=42,
    duration_ms=1200,
    prompt='What is the capital of France?',
    response='The capital of France is Paris.'
)

# Send a heartbeat
agent.heartbeat()

# Cleanup
agent.shutdown()

API Reference

Constructor

agent = NexusAgent(
    api_key,           # (required) API key for NexusOS
    agent_id,          # (required) Agent ID
    base_url='https://nexusos-backend-7ue5.onrender.com',  # (optional) Base API URL
    batch_size=10,     # (optional) Number of logs to batch before flushing
    flush_interval=5   # (optional) Interval in seconds to flush logs
)

Methods

log(payload)

Add a generic log event to the batch queue. Automatically flushes when batch reaches batch_size.

agent.log({
    'event': 'custom.event',
    'data': { 'key': 'value' }
})

log_llm_call(model, tokens_used, duration_ms, prompt=None, response=None)

Log a large language model call.

agent.log_llm_call(
    model='gpt-4',
    tokens_used=350,
    duration_ms=2500,
    prompt='Summarize this article...',
    response='The article discusses...'
)

log_http_request(url, method, status, duration_ms, success)

Log an HTTP request.

agent.log_http_request(
    url='https://api.example.com/data',
    method='GET',
    status=200,
    duration_ms=450,
    success=True
)

log_action(tool, input=None, output=None, duration_ms=None, success=True, error_message=None)

Log a tool action execution.

agent.log_action(
    tool='web_search',
    input={'query': 'node.js best practices'},
    output={'results': [...]},
    duration_ms=800,
    success=True
)

heartbeat()

Send a heartbeat signal to indicate the agent is alive.

agent.heartbeat()

is_alive()

Check if the agent is alive by fetching its stats from the API.

alive = agent.is_alive()
print(f'Agent is alive: {alive}')

start_heartbeat(interval=60)

Start a background thread to send periodic heartbeats.

agent.start_heartbeat(interval=60)

trace(name)

Context manager to trace the execution of a code block.

with agent.trace('process_data') as t:
    result = process_data()
    t.set_output(result)

monitor(tool=None)

Decorator to automatically log function execution.

@agent.monitor(tool='web_search')
def search_web(query):
    return results

# Works with async functions too
@agent.monitor(tool='fetch_data')
async def fetch_data(url):
    return await get_data(url)

shutdown()

Shutdown the agent: stop all threads and flush remaining logs.

agent.shutdown()

Context Manager

Use the agent as a context manager for automatic cleanup:

with NexusAgent(api_key='nxs_live_xxxx', agent_id='nexus_agt_xxxx') as agent:
    agent.log_llm_call(model='gpt-4', tokens_used=100, duration_ms=500)
    # shutdown() is called automatically

Sensitive Data Handling

The SDK automatically sanitizes logs to remove sensitive information. Keys containing the following terms (case-insensitive) are redacted:

• password
• secret
• key
• token
• auth

Example:

agent.log({
    'username': 'john_doe',
    'password': 'super_secret',
    'api_key': 'sk_live_xxx'
})

# Logged as:
# {
#     'username': 'john_doe',
#     'password': '[REDACTED]',
#     'api_key': '[REDACTED]'
# }

Decorator Style

Log function execution with the @monitor decorator:

from nexusos import NexusAgent

agent = NexusAgent(
    api_key='nxs_live_xxxx',
    agent_id='nexus_agt_xxxx'
)

@agent.monitor(tool='calculate_sum')
def calculate_sum(numbers):
    return sum(numbers)

result = calculate_sum([1, 2, 3])
# Automatically logs the function execution with timing and success/error

Context Manager Style

Use the trace() context manager for more control:

with agent.trace('data_processing') as trace:
    processed_data = process_data()
    trace.set_output(processed_data)

# Automatically logs timing, duration, and output

Kill Switch & Pause — IMPORTANT

The NexusOS dashboard gives you a Kill and Pause button for every agent. However, these only work if your agent actually calls is_alive() in its main loop and respects the result.

Without is_alive(), your agent will keep running even after you kill or pause it from the dashboard — logs will just start failing silently.

The Pattern

Wrap your main cycle with an is_alive() check at the top:

import os
import time
from nexusos import NexusAgent

agent = NexusAgent(
    api_key=os.getenv('NEXUSOS_API_KEY'),
    agent_id=os.getenv('NEXUSOS_AGENT_ID'),
)

def run_cycle():
    # ✅ REQUIRED — check kill/pause status before doing any work
    if not agent.is_alive():
        print('Agent paused or killed from NexusOS dashboard — skipping cycle')
        return

    # your agent logic here...
    agent.heartbeat()

# Run on an interval
while True:
    run_cycle()
    time.sleep(5 * 60)  # 5 minutes

What each status does

Dashboard Action	is_alive() returns	Effect
Active (default)	True	Agent runs normally
Pause	False	Cycle is skipped, resumes when you unpause
Kill	False	Cycle is skipped permanently until process restart

Note: is_alive() makes a lightweight GET request to NexusOS. Call it once per cycle, not on every log line.

LangChain Integration

Integrate NexusOS with LangChain to automatically monitor LLM calls:

from nexusos import NexusAgent
from langchain.llms import OpenAI
from langchain.agents import initialize_agent, load_tools

agent = NexusAgent(
    api_key='nxs_live_xxxx',
    agent_id='nexus_agt_xxxx'
)

# Start periodic heartbeats
agent.start_heartbeat(interval=60)

llm = OpenAI(temperature=0)

# Wrap the LLM call
original_generate = llm.generate

def monitored_generate(prompts, *args, **kwargs):
    import time
    start = time.time()
    try:
        result = original_generate(prompts, *args, **kwargs)
        duration_ms = int((time.time() - start) * 1000)
        agent.log_llm_call(
            model=llm.model_name,
            tokens_used=result.llm_output.get('token_usage', {}).get('total_tokens', 0),
            duration_ms=duration_ms,
            prompt=str(prompts),
            response=result.generations[0][0].text
        )
        return result
    except Exception as e:
        duration_ms = int((time.time() - start) * 1000)
        agent.log_llm_call(
            model=llm.model_name,
            tokens_used=0,
            duration_ms=duration_ms,
            prompt=str(prompts),
            response=f'Error: {str(e)}'
        )
        raise

llm.generate = monitored_generate

tools = load_tools(['serpapi'])
agent_executor = initialize_agent(tools, llm, agent='zero-shot-react-description')

result = agent_executor.run('What is the capital of France?')
agent.shutdown()

OpenAI Integration

Monitor OpenAI API calls directly:

from nexusos import NexusAgent
import openai
import time

agent = NexusAgent(
    api_key='nxs_live_xxxx',
    agent_id='nexus_agt_xxxx'
)

def call_openai(prompt):
    start = time.time()
    try:
        response = openai.ChatCompletion.create(
            model='gpt-4',
            messages=[{'role': 'user', 'content': prompt}]
        )
        duration_ms = int((time.time() - start) * 1000)
        tokens_used = response['usage']['total_tokens']

        agent.log_llm_call(
            model='gpt-4',
            tokens_used=tokens_used,
            duration_ms=duration_ms,
            prompt=prompt,
            response=response['choices'][0]['message']['content']
        )

        return response
    except Exception as e:
        duration_ms = int((time.time() - start) * 1000)
        agent.log_llm_call(
            model='gpt-4',
            tokens_used=0,
            duration_ms=duration_ms,
            prompt=prompt,
            response=f'Error: {str(e)}'
        )
        raise

result = call_openai('What is the capital of France?')
agent.shutdown()

Environment Variables

Configure the SDK using environment variables:

export NEXUSOS_API_KEY=nxs_live_xxxx
export NEXUSOS_AGENT_ID=nexus_agt_xxxx
export NEXUSOS_BASE_URL=https://nexusos-backend-7ue5.onrender.com

import os
from nexusos import NexusAgent

agent = NexusAgent(
    api_key=os.getenv('NEXUSOS_API_KEY'),
    agent_id=os.getenv('NEXUSOS_AGENT_ID'),
    base_url=os.getenv('NEXUSOS_BASE_URL', 'https://nexusos-backend-7ue5.onrender.com')
)

Async Support

The @monitor decorator and trace() context manager fully support async functions:

@agent.monitor(tool='async_fetch')
async def fetch_data(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.json()

result = await fetch_data('https://api.example.com/data')

License

MIT