OpenClaw + Syncreus Integration Guide

Connect quality monitoring to your OpenClaw agents in under 10 minutes

Works with NemoClaw for the complete safety + observability stack

Prerequisites

• ✓A Syncreus account (app.syncreus.com)
• ✓An OpenClaw project (new or existing)
• ✓Python 3.9+

Add the Syncreus SDK to your OpenClaw project:

pip install syncreus-sdk

Or if using the OpenClaw project template:

cd my-openclaw-project
pip install -r requirements.txt syncreus-sdk

Add to your .env file (never commit this file):

# .env
SYNCREUS_API_KEY=syn_live_xxxxxxxxxxxxxxxx
SYNCREUS_PROJECT=my-openclaw-agent
SYNCREUS_ENVIRONMENT=production

Or if deploying to a cloud platform, add SYNCREUS_API_KEY to your environment variables in the same place you set OPENCLAW_API_KEY.

OpenClaw tools are Python functions decorated with @tool. Wrap them with @syncreus.trace to monitor every tool invocation:

import syncreus
import os
import requests
from openclaw import Agent, tool, run_agent

syncreus.init(
    api_key=os.environ["SYNCREUS_API_KEY"],
    project=os.environ.get("SYNCREUS_PROJECT", "openclaw-agent"),
    environment=os.environ.get("SYNCREUS_ENVIRONMENT", "production"),
)


@tool
@syncreus.trace(name="search_web", tags={"tool_type": "search"})
def search_web(query: str) -> str:
    """Search the web and return a summary of results."""
    response = requests.get(
        "https://api.search.example.com/search",
        params={"q": query, "limit": 5},
        headers={"Authorization": f"Bearer {os.environ['SEARCH_API_KEY']}"},
    )
    results = response.json()["results"]
    return "\n".join(f"- {r['title']}: {r['snippet']}" for r in results)


@tool
@syncreus.trace(name="write_file", tags={"tool_type": "filesystem"})
def write_file(path: str, content: str) -> str:
    """Write content to a file."""
    with open(path, "w") as f:
        f.write(content)
    return f"Successfully wrote {len(content)} characters to {path}"


@tool
@syncreus.trace(name="execute_python", tags={"tool_type": "code_execution"})
def execute_python(code: str) -> str:
    """Execute Python code in a sandboxed subprocess and return the output."""
    import subprocess
    result = subprocess.run(
        ["python", "-c", code],
        capture_output=True,
        text=True,
        timeout=30,
    )
    return result.stdout or result.stderr

OpenClaw agents run continuously, which makes loop detection and budget enforcement critical. Configure both with the Syncreus agent monitor:

from syncreus.openclaw import SyncreusAgentMonitor, LoopDetector, BudgetGuard
from openclaw import Agent, run_agent

monitor = SyncreusAgentMonitor(
    session_name="production_research_agent",
    metadata={
        "agent_version": "2.1.0",
        "deployment": "railway",
    },
)

loop_detector = LoopDetector(
    monitor=monitor,
    # Alert if the agent repeats the same action 3 times in a row.
    max_repetitions=3,
    similarity_threshold=0.88,
    on_loop="alert_and_stop",
    alert_channels=["whatsapp", "email"],  # Configured in Syncreus dashboard.
)

budget_guard = BudgetGuard(
    monitor=monitor,
    max_usd=5.00,
    warn_at_usd=4.00,
    on_exceed="stop",
)

Here is a full competitive research agent with all Syncreus monitoring components wired together:

import os
import syncreus
import requests
from syncreus.openclaw import SyncreusAgentMonitor, LoopDetector, BudgetGuard
from openclaw import Agent, tool, run_agent

# Initialize Syncreus.
syncreus.init(
    api_key=os.environ["SYNCREUS_API_KEY"],
    project="competitive-research-agent",
    environment="production",
)

monitor = SyncreusAgentMonitor(session_name="competitive_analysis")

loop_detector = LoopDetector(
    monitor=monitor,
    max_repetitions=3,
    similarity_threshold=0.88,
    on_loop="alert_and_stop",
    alert_channels=["whatsapp"],
)

budget_guard = BudgetGuard(
    monitor=monitor,
    max_usd=2.00,
    warn_at_usd=1.50,
    on_exceed="stop",
)


@tool
@syncreus.trace(name="search_competitor", tags={"tool_type": "research"})
def search_competitor(company_name: str) -> str:
    """Search for information about a competitor company."""
    response = requests.get(
        "https://api.search.example.com/search",
        params={"q": f"{company_name} product features pricing 2026"},
        headers={"Authorization": f"Bearer {os.environ['SEARCH_API_KEY']}"},
        timeout=10,
    )
    results = response.json()["results"]
    return "\n".join(f"- {r['title']}: {r['snippet']}" for r in results[:5])


@tool
@syncreus.trace(name="scrape_pricing_page", tags={"tool_type": "scraping"})
def scrape_pricing_page(url: str) -> str:
    """Scrape pricing information from a URL."""
    response = requests.get(url, timeout=10, headers={"User-Agent": "Mozilla/5.0"})
    return response.text[:3000]


@tool
@syncreus.trace(name="save_report", tags={"tool_type": "output"})
def save_report(filename: str, content: str) -> str:
    """Save the research report to a file."""
    with open(filename, "w") as f:
        f.write(content)
    return f"Report saved to {filename} ({len(content)} characters)"


# Build the agent.
agent = Agent(
    name="Competitive Intelligence Agent",
    model="gpt-4o",
    instructions="""
    You are a competitive intelligence researcher. Your job is to:
    1. Research the given competitors using the search and scraping tools.
    2. Synthesize findings into a structured comparison report.
    3. Save the report as a Markdown file.

    Be thorough but efficient. Do not repeat searches you have already done.
    """,
    tools=[search_competitor, scrape_pricing_page, save_report],
)

# Register Syncreus monitoring components.
monitor.register(agent)
loop_detector.attach(agent)
budget_guard.attach(agent)

# Run the agent.
try:
    result = run_agent(
        agent=agent,
        task="Research our top 3 competitors: Acme AI, BetaCorp, and Gamma Systems. "
             "Compare their pricing, key features, and recent product announcements. "
             "Save the results to competitive_analysis_2026.md",
    )
    print("Agent completed successfully.")
    print(f"Total cost: ${monitor.get_session_cost():.4f}")
except BudgetGuard.BudgetExceededError as e:
    print(f"Agent stopped by budget guard: {e}")
except LoopDetector.LoopDetectedError as e:
    print(f"Agent stopped by loop detector: {e}")

Adding NemoClaw for Policy Enforcement

NemoClaw is NVIDIA's enterprise plugin for OpenClaw. It adds policy-based guardrails to OpenClaw agents — defining what the agent is allowed to do, which APIs it may call, which domains it may browse, and which actions require human approval.

Running both NemoClaw and Syncreus gives you the complete agent safety and quality stack: NemoClaw ensures the agent stays within policy boundaries; Syncreus ensures the work it does within those boundaries is actually good.

NemoClaw vs Syncreus

NemoClaw + Syncreus Configuration

import os
import syncreus
from syncreus.openclaw import SyncreusAgentMonitor, LoopDetector, BudgetGuard
from openclaw import Agent, tool, run_agent
from nemoclaw import NemoClawPlugin, PolicyConfig
import requests

# Initialize Syncreus.
syncreus.init(
    api_key=os.environ["SYNCREUS_API_KEY"],
    project="nemoclaw-enterprise-agent",
    environment="production",
)

monitor = SyncreusAgentMonitor(session_name="policy_governed_agent")
loop_detector = LoopDetector(monitor=monitor, max_repetitions=3, on_loop="alert_and_stop")
budget_guard = BudgetGuard(monitor=monitor, max_usd=10.00, on_exceed="stop")

# Configure NemoClaw policy.
nemoclaw = NemoClawPlugin(
    policy=PolicyConfig(
        allowed_domains=["*.company.com", "*.approved-vendor.com"],
        blocked_actions=["delete_file", "send_email_external"],
        require_approval_for=["publish_report", "update_database"],
        max_tool_calls_per_run=50,
    ),
    nvidia_api_key=os.environ["NVIDIA_API_KEY"],
)


@tool
@syncreus.trace(name="fetch_internal_data", tags={"tool_type": "data_retrieval"})
@nemoclaw.guard  # NemoClaw checks policy before Syncreus logs the call.
def fetch_internal_data(endpoint: str) -> str:
    """Fetch data from an approved internal API endpoint."""
    response = requests.get(
        f"https://api.company.com/{endpoint}",
        headers={"Authorization": f"Bearer {os.environ['INTERNAL_API_KEY']}"},
        timeout=10,
    )
    return str(response.json())


@tool
@syncreus.trace(name="publish_report", tags={"tool_type": "output"})
@nemoclaw.guard  # NemoClaw will pause here and request human approval.
def publish_report(title: str, content: str) -> str:
    """Publish a report to the internal dashboard. Requires human approval."""
    return f"Report '{title}' published successfully."


agent = Agent(
    name="Enterprise Data Analysis Agent",
    model="meta/llama-3.1-70b-instruct",  # Running on NVIDIA NIM
    instructions="""
    You are an enterprise data analysis agent. You must:
    1. Only access approved internal data sources.
    2. Always request human approval before publishing any report.
    3. Stay within the allowed domain list.
    """,
    tools=[fetch_internal_data, publish_report],
)

# Register both NemoClaw and Syncreus.
nemoclaw.register(agent)
monitor.register(agent)
loop_detector.attach(agent)
budget_guard.attach(agent)

# Run — NemoClaw enforces policy, Syncreus monitors quality.
result = run_agent(
    agent=agent,
    task="Analyze Q1 2026 sales data from the internal API and publish a summary report.",
)

How the layers work together at runtime

User Task
    |
    v
OpenClaw Agent decides to call a tool
    |
    v
NemoClaw policy check:
  |-- Is this action allowed? --> No --> Block + log policy violation
  +-- Yes --> Is approval required? --> Yes --> Pause + notify approver
                                   --> No --> Allow execution
    |
    v
Tool executes
    |
    v
Syncreus captures:
  |-- Tool name + inputs
  |-- Tool output
  |-- Latency
  |-- Quality score
  +-- Loop detection check
    |
    v
Syncreus dashboard -- full trace with policy events overlaid

What You Get with OpenClaw + Syncreus

Troubleshooting

@tool
@syncreus.trace(name="my_tool")   # Outer -- logs the full round-trip
@nemoclaw.guard                    # Inner -- policy check before execution
def my_tool(...):
    ...