Tutorials

Tutorial 1: Smart Home Agent

In this tutorial you will build a smart home controller that manages a front door lock, living room lights, and a thermostat. The agent accepts natural-language commands, parses them locally with OmniLinkEngine, executes device logic, and reports results through AgentFeedback.

Prerequisites

• Python 3.10 or later
• An Omni Key (omk_...) from the OmniLink Dashboard
• pip install omnilink (or pip install -e omnilink-lib/ from the repo)

Step 1: Define the Command Engine

Start by creating an OmniLinkEngine instance and registering custom types for the devices your agent will control.

from omnilink import OmniLinkEngine, AgentFeedback

engine = OmniLinkEngine()

# Register custom types for strict validation
engine.register_type("lock_state", ["locked", "unlocked"])
engine.register_type("light_state", ["on", "off"])
engine.register_type("room", ["living room", "bedroom", "kitchen", "bathroom"])

Step 2: Create Command Handlers

Each command template maps a natural-language pattern to a Python function. The engine extracts typed variables automatically.

# ── Door lock ──
@engine.command("set front door to [state:lock_state]")
def set_door(state: str):
    """Lock or unlock the front door."""
    # In production, this would call your smart lock API
    feedback = AgentFeedback()
    feedback.message = f"Front door is now {state}."
    feedback.success = True
    return feedback

# ── Lights ──
@engine.command("turn [room:room] lights [state:light_state]")
def set_lights(room: str, state: str):
    """Control lights in a specific room."""
    feedback = AgentFeedback()
    feedback.message = f"{room.title()} lights turned {state}."
    feedback.success = True
    return feedback

# ── Thermostat ──
@engine.command("set thermostat to [temp:int] degrees")
def set_thermostat(temp: int):
    """Set the target temperature."""
    if temp < 60 or temp > 85:
        feedback = AgentFeedback()
        feedback.message = f"Temperature {temp}°F is out of safe range (60-85)."
        feedback.success = False
        return feedback
    feedback = AgentFeedback()
    feedback.message = f"Thermostat set to {temp}°F."
    feedback.success = True
    return feedback

Step 3: Connect the HTTP Bridge

The HTTP bridge exposes your engine on a local REST endpoint so external systems (or the OmniLink cloud) can send commands to it.

from omnilink import OmniLinkHTTPBridge

bridge = OmniLinkHTTPBridge(engine, port=8080)

# This starts a Flask server on http://localhost:8080
# POST /command with {"text": "turn living room lights on"}
bridge.start()

Step 4: Register on the Platform

To let the cloud AI dispatch commands to your agent, register it with the OmniLink platform using your Omni Key.

from omnilink import OmniLinkClient
import os

client = OmniLinkClient(omni_key=os.environ["OMNI_KEY"])

# List available commands so the AI knows what your agent can do
commands = engine.get_available_commands()
print("Registered commands:")
for cmd in commands:
    print(f"  {cmd}")

Step 5: Test the Agent

# Start the agent
export OMNI_KEY="omk_your_key_here"
python smart_home.py

# In another terminal, send a command
curl -X POST http://localhost:8080/command \
  -H "Content-Type: application/json" \
  -d '{"text": "set front door to locked"}'

# Response:
# {"success": true, "message": "Front door is now locked."}

Full Source Code

"""smart_home.py — OmniLink Smart Home Agent"""
import os
from omnilink import (
    OmniLinkEngine,
    OmniLinkHTTPBridge,
    OmniLinkClient,
    AgentFeedback,
)

# ── Engine setup ──
engine = OmniLinkEngine()
engine.register_type("lock_state", ["locked", "unlocked"])
engine.register_type("light_state", ["on", "off"])
engine.register_type("room", ["living room", "bedroom", "kitchen", "bathroom"])

@engine.command("set front door to [state:lock_state]")
def set_door(state: str):
    feedback = AgentFeedback()
    feedback.message = f"Front door is now {state}."
    feedback.success = True
    return feedback

@engine.command("turn [room:room] lights [state:light_state]")
def set_lights(room: str, state: str):
    feedback = AgentFeedback()
    feedback.message = f"{room.title()} lights turned {state}."
    feedback.success = True
    return feedback

@engine.command("set thermostat to [temp:int] degrees")
def set_thermostat(temp: int):
    if temp < 60 or temp > 85:
        feedback = AgentFeedback()
        feedback.message = f"Temperature {temp}°F is out of safe range (60-85)."
        feedback.success = False
        return feedback
    feedback = AgentFeedback()
    feedback.message = f"Thermostat set to {temp}°F."
    feedback.success = True
    return feedback

# ── Connect ──
client = OmniLinkClient(omni_key=os.environ["OMNI_KEY"])
bridge = OmniLinkHTTPBridge(engine, port=8080)

print("Smart Home Agent running on http://localhost:8080")
print(f"Available commands: {engine.get_available_commands()}")
bridge.start()

Tutorial 2: Game-Playing Agent with ToolRunner

This tutorial demonstrates the ToolRunner lifecycle by building an agent that navigates a simple grid world. The AI decides the strategy; the local runtime executes each move. A full 30-minute session costs only 1–2 credits.

Understanding the ToolRunner Lifecycle

1. Kickoff (1 credit) — The cloud AI receives the game state and decides on an action plan.
2. Local loop (0 credits) — Your code calls execute_action(), updates state, and checks for game over.
3. Optional periodic reviews — If enabled, the AI re-evaluates progress at intervals.
4. Final analysis (1 credit) — The AI summarizes the session outcome.

Step 1: Define the Game World

GRID_SIZE = 5

# Simple grid: player starts at (0,0), goal at (4,4)
# Obstacles at fixed positions
OBSTACLES = {(1, 1), (2, 3), (3, 1), (3, 2)}

Step 2: Subclass ToolRunner

from omnilink import ToolRunner
import os

class GridNavigator(ToolRunner):
    def __init__(self):
        super().__init__(
            omni_key=os.environ["OMNI_KEY"],
            tool_name="grid-navigator",
            description="Navigate a 5x5 grid from start to goal while avoiding obstacles.",
        )
        self.player = [0, 0]
        self.goal = [4, 4]
        self.moves = 0
        self.max_moves = 20
        self.history = []

    def get_state(self) -> dict:
        """Return current game state for the AI to evaluate."""
        return {
            "player_position": self.player,
            "goal_position": self.goal,
            "obstacles": [list(o) for o in OBSTACLES],
            "grid_size": GRID_SIZE,
            "moves_taken": self.moves,
            "moves_remaining": self.max_moves - self.moves,
            "move_history": self.history[-5:],  # last 5 moves
        }

    def execute_action(self, action: str) -> str:
        """Execute a movement action. Returns result message."""
        direction_map = {
            "up": (0, -1),
            "down": (0, 1),
            "left": (-1, 0),
            "right": (1, 0),
        }

        action = action.strip().lower()
        if action not in direction_map:
            return f"Invalid action: {action}. Use up, down, left, or right."

        dx, dy = direction_map[action]
        new_x = self.player[0] + dx
        new_y = self.player[1] + dy

        # Boundary check
        if not (0 <= new_x < GRID_SIZE and 0 <= new_y < GRID_SIZE):
            self.moves += 1
            self.history.append(f"{action} (blocked by wall)")
            return f"Cannot move {action}: hit the grid boundary."

        # Obstacle check
        if (new_x, new_y) in OBSTACLES:
            self.moves += 1
            self.history.append(f"{action} (blocked by obstacle)")
            return f"Cannot move {action}: obstacle at ({new_x}, {new_y})."

        # Valid move
        self.player = [new_x, new_y]
        self.moves += 1
        self.history.append(f"{action} -> ({new_x}, {new_y})")
        return f"Moved {action} to ({new_x}, {new_y}). Moves: {self.moves}/{self.max_moves}"

    def state_summary(self) -> str:
        """Human-readable summary for periodic reviews."""
        return (
            f"Position: ({self.player[0]}, {self.player[1]}) | "
            f"Goal: ({self.goal[0]}, {self.goal[1]}) | "
            f"Moves: {self.moves}/{self.max_moves}"
        )

    def is_game_over(self) -> bool:
        """Check if the game has ended."""
        if self.player == self.goal:
            print("Goal reached!")
            return True
        if self.moves >= self.max_moves:
            print("Out of moves!")
            return True
        return False

Step 3: Run the Agent

if __name__ == "__main__":
    navigator = GridNavigator()
    navigator.run()

Running It

export OMNI_KEY="omk_your_key_here"
python grid_navigator.py

# Output:
# [ToolRunner] Starting grid-navigator...
# [ToolRunner] Kickoff: AI analyzing initial state (1 credit)
# [ToolRunner] Executing: right -> (1, 0). Moves: 1/20
# [ToolRunner] Executing: down -> (1, 1) blocked by obstacle
# [ToolRunner] Executing: right -> (2, 0). Moves: 3/20
# ...
# [ToolRunner] Goal reached!
# [ToolRunner] Final analysis (1 credit): Navigated to goal in 9 moves.
# [ToolRunner] Total credits used: 2

Credit Breakdown

Tutorial 3: Customer Support Bot with Knowledge

Build a customer support agent that answers questions grounded in your documentation. This tutorial covers knowledge upload, deep knowledge mode, and chat integration in both Python and TypeScript.

Step 1: Upload Knowledge Files

First, upload your support documentation to the OmniLink knowledge base. The platform automatically chunks, embeds, and indexes the content for retrieval-augmented generation.

from omnilink import OmniLinkClient
import os

client = OmniLinkClient(omni_key=os.environ["OMNI_KEY"])

# Upload support documentation
client.upload_knowledge(
    file_path="docs/product-manual.pdf",
    metadata={"category": "product", "version": "2.1"}
)

client.upload_knowledge(
    file_path="docs/troubleshooting-guide.md",
    metadata={"category": "troubleshooting"}
)

client.upload_knowledge(
    file_path="docs/pricing-faq.txt",
    metadata={"category": "billing"}
)

print("Knowledge files uploaded and indexed.")

Step 2: Configure the Agent Profile

Set up an agent profile with system instructions tailored for customer support. Enable deep knowledge mode so the AI retrieves the most relevant snippets before generating a response.

# Configure the agent for support
client.update_agent_profile(
    system_instructions="""You are a helpful customer support agent for Acme Corp.
Always be polite, professional, and concise.
If you do not know the answer, say so honestly and offer to escalate.
Use the knowledge base to ground your responses in official documentation.
Never make up product features or pricing.""",
    deep_knowledge=True,
    engine="g4-engine",  # Claude for nuanced support interactions
)

Step 3: Chat with Knowledge (Python)

# Send a support query
response = client.chat(
    message="How do I reset my device to factory settings?",
    use_knowledge=True,
)

print(f"Agent: {response['message']}")
print(f"Sources: {response.get('knowledge_sources', [])}")

Step 4: Multi-Turn Conversations with Memory

OmniLink automatically maintains short-term memory within a conversation. For persistent context, use long-term memory.

# The agent remembers previous messages in the conversation
response1 = client.chat(message="I bought the Pro plan last week.")
response2 = client.chat(message="Can I upgrade to Diamond?")
# The agent knows the user is on Pro from the previous message

# Store important context in long-term memory
client.save_long_term_memory(
    key="customer_plan",
    value="Pro plan, purchased 2026-03-17"
)

Production Tips

• Use g4-engine (Claude) for nuanced, empathetic support responses.
• Use g1-engine (Gemini) for high-volume, cost-efficient FAQ handling.
• Enable deep knowledge mode to ensure responses are grounded in your documentation.
• Set clear system instructions that define tone, boundaries, and escalation paths.
• Monitor usage to avoid unexpected credit consumption during traffic spikes.

Tutorial 4: Robot Mission Controller

Build a robot controller that uses the HTTP bridge for bidirectional communication. The agent receives mission commands, dispatches movement and sensor instructions, and sends real-time progress updates back to the cloud.

Prerequisites

• Python 3.10 or later
• A local agent HTTP server running on localhost:8080
• pip install omnilink (or pip install -e omnilink-lib/ from the repo)

Step 1: Define Movement and Sensor Commands

from omnilink import OmniLinkEngine, AgentFeedback

engine = OmniLinkEngine()

# Register custom types
engine.register_type("direction", ["forward", "backward", "left", "right"])
engine.register_type("sensor", ["lidar", "camera", "imu", "ultrasonic"])

# ── Movement commands ──
@engine.command("move [direction:direction] [distance:float] meters")
def move(direction: str, distance: float):
    feedback = AgentFeedback()
    feedback.message = f"Moving {direction} {distance}m."
    feedback.success = True
    feedback.data = {
        "command": "move",
        "direction": direction,
        "distance": distance,
        "status": "executing",
    }
    return feedback

@engine.command("rotate [angle:int] degrees")
def rotate(angle: int):
    feedback = AgentFeedback()
    feedback.message = f"Rotating {angle} degrees."
    feedback.success = True
    feedback.data = {"command": "rotate", "angle": angle, "status": "executing"}
    return feedback

@engine.command("stop")
def stop():
    feedback = AgentFeedback()
    feedback.message = "Robot stopped."
    feedback.success = True
    feedback.data = {"command": "stop", "status": "idle"}
    return feedback

# ── Sensor commands ──
@engine.command("read [sensor:sensor] sensor")
def read_sensor(sensor: str):
    # In production, read from actual hardware
    sensor_data = {
        "lidar": {"ranges": [1.2, 0.8, 2.5, 3.1], "unit": "meters"},
        "camera": {"frame_id": 1042, "resolution": "640x480"},
        "imu": {"roll": 0.02, "pitch": -0.01, "yaw": 45.3},
        "ultrasonic": {"distance": 0.75, "unit": "meters"},
    }
    feedback = AgentFeedback()
    feedback.message = f"Sensor data from {sensor}: {sensor_data[sensor]}"
    feedback.success = True
    feedback.data = sensor_data[sensor]
    return feedback

@engine.command("set speed to [speed:int] percent")
def set_speed(speed: int):
    if speed < 0 or speed > 100:
        feedback = AgentFeedback()
        feedback.message = f"Speed {speed}% is out of range (0-100)."
        feedback.success = False
        return feedback
    feedback = AgentFeedback()
    feedback.message = f"Speed set to {speed}%."
    feedback.success = True
    return feedback

Step 2: Set Up the HTTP Bridge

The HTTP bridge exposes POST /command for incoming commands and GET /feedback for retrieving results.

from omnilink import OmniLinkHTTPBridge

bridge = OmniLinkHTTPBridge(
    engine,
    host="0.0.0.0",
    port=8080,
)

Step 3: Send Progress Updates

During long-running missions, publish progress updates so the cloud AI and dashboard can track the robot in real time.

import json
import time

def execute_mission(bridge, waypoints):
    """Navigate through a list of waypoints with progress reporting."""
    total = len(waypoints)
    for i, (x, y) in enumerate(waypoints):
        # Execute movement
        result = engine.parse(f"move forward {x} meters")
        print(f"Waypoint {i+1}/{total}: {result.message}")

        # Publish progress update
        progress = {
            "type": "progress",
            "waypoint": i + 1,
            "total_waypoints": total,
            "position": {"x": x, "y": y},
            "percent_complete": round((i + 1) / total * 100),
        }
        bridge.publish(json.dumps(progress))

        time.sleep(1)  # Simulate movement time

    # Mission complete
    bridge.publish(json.dumps({
        "type": "mission_complete",
        "waypoints_visited": total,
    }))

Step 4: Handle Bidirectional Messaging

The HTTP bridge supports incoming commands from external systems. You can add a callback for custom message handling.

def on_external_message(topic, payload):
    """Handle messages from external systems (e.g., a fleet manager)."""
    data = json.loads(payload)
    if data.get("type") == "emergency_stop":
        engine.parse("stop")
        print("Emergency stop triggered by external system.")
    elif data.get("type") == "new_mission":
        waypoints = data["waypoints"]
        execute_mission(bridge, waypoints)

bridge.on_message = on_external_message

Step 5: Run the Complete Controller

"""robot_controller.py — OmniLink Robot Mission Controller"""
import os

# Start the HTTP bridge (blocks and listens for commands)
print("Robot Mission Controller running.")
print(f"Listening on: POST /command")
print(f"Feedback at: GET /feedback")
print(f"Available commands: {engine.get_available_commands()}")

bridge.start()

Testing with curl

# Terminal 1: Start the controller
export OMNI_KEY="omk_your_key_here"
python robot_controller.py

# Terminal 2: Get feedback from the agent
curl http://localhost:8080/feedback

# Terminal 2: Get current agent context
curl http://localhost:8080/context

# Terminal 2: Send a command
curl -X POST http://localhost:8080/command \
  -H "Content-Type: application/json" \
  -d '{"text": "move forward 2.5 meters"}'

# Terminal 2: Send a mission
curl -X POST http://localhost:8080/command \
  -H "Content-Type: application/json" \
  -d '{"type": "new_mission", "waypoints": [[1,0],[2,1],[3,2],[4,4]]}'

Next Steps

Now that you have built working agents, explore these resources to take your projects further:

• Security — Harden your agents for production deployment.
• Usage Management — Monitor and optimize your credit consumption.
• API Reference — Complete endpoint documentation.
• FAQ — Common questions and answers.