omnilink 0.4.3

OmniLink natural language engine and integration bridges

OmniLink Python Library (v0.4.3)

OmniLink is a Python toolkit for building AI-powered automation agents. It provides two complementary systems:

• ToolRunner — Run AI controllers locally (games, robots, any task) with native structured tool calling via each AI provider's API (OpenAI function calling, Anthropic tool_use, Gemini functionDeclarations).
• Command Engine — Match natural-language commands to Python handlers with variable extraction, bidirectional messaging, and HTTP REST transport.

---

Table of Contents

1. Installation
2. Get your Omni Key
3. Quick start — ToolRunner
4. ToolRunner reference
5. Built-in benchmark examples
6. Core concepts — command engine
7. Quick start — engine only
8. Templates and types
9. Writing handlers
10. Bidirectional messaging
11. Connecting over HTTP
12. REST API client
13. Environment variable reference
14. Running the examples
15. Running the tests

---

Installation

# From PyPI (when published)
pip install omnilink

# Or install from source (editable)
git clone https://github.com/omnilink/omnilink
pip install -e omnilink/omnilink-lib

Python 3.9 or later is required. All common dependencies (requests) are installed automatically. The Chess example additionally requires the chess package (pip install chess).

---

Get Your Omni Key

Before using the library you need an Omni Key (olink_...). Generate one with a single click:

Get your Omni Key

Sign in (or create a free account) and click Generate API key. Copy the key — you will need it for every API call.

---

Quick start — ToolRunner

A ToolRunner is a Python controller that runs on your machine and is orchestrated by OmniLink's cloud AI. The cloud agent triggers a tool call (e.g. make_move), which hands control to your local code. Your code runs the actual logic — game AI, robot navigation, data pipeline — at full speed with zero per-action API calls.

How it works

┌─────────────────┐    1 API call     ┌──────────────────┐
│  OmniLink Cloud │ ───────────────── │  Your ToolRunner │
│  (Chat + Memory)│   "Call make_move"│  (local Python)  │
└─────────────────┘                   └────────┬─────────┘
                                               │ polls state,
                                               │ sends actions
                                      ┌────────▼─────────┐
                                      │   Target System   │
                                      │ (game server,     │
                                      │  robot, API, etc.)│
                                      └──────────────────┘

Credit usage: 1 credit to kick off + 1 for final analysis. Periodic reviews are optional. A 30-minute session typically costs 1–2 credits total.

Set your Omni Key

ToolRunner reads your key from the OMNI_KEY environment variable:

export OMNI_KEY="olink_YOUR_KEY_HERE"

You can also override it per-runner by setting omni_key as a class attribute.

Minimal example

from omnilink.tool_runner import ToolRunner

class MyRunner(ToolRunner):
    agent_name = "my-agent"
    display_name = "My Task"

    def get_state(self):
        # Fetch state from your target system (HTTP, socket, file, etc.)
        return requests.get("http://localhost:8000/state").json()

    def execute_action(self, state):
        # Decide what to do and send the action
        action = "UP" if state["y"] < state["target_y"] else "DOWN"
        requests.post("http://localhost:8000/action", json={"action": action})

    def state_summary(self, state):
        # Concise text for the agent's memory
        return f"Position: ({state['x']}, {state['y']}), Target: ({state['target_x']}, {state['target_y']})"

    def is_game_over(self, state):
        return state.get("done", False)

if __name__ == "__main__":
    MyRunner().run()

That's it — python my_runner.py will:

1. Create/update the agent profile on OmniLink
2. Ask the cloud agent to call make_move (1 API credit)
3. Run your local loop: poll state → execute action → repeat
4. Persist state to agent memory every 60 seconds
5. Listen for pause/resume/stop commands from the OmniLink UI
6. Print a final summary and ask the agent for analysis (1 API credit)

Real-world example (Breakout)

from omnilink.tool_runner import ToolRunner
from .breakout_api import get_state, send_action, SERVER_URL
from .breakout_engine import decide_action, state_summary

class BreakoutRunner(ToolRunner):
    agent_name = "breakout-agent"
    display_name = "Breakout"
    tool_description = "Move paddle."
    game_server_url = SERVER_URL

    def __init__(self):
        self._last_score = 0

    def get_state(self):
        return get_state()

    def execute_action(self, state):
        if state.get("game_state") == "PLAY":
            send_action(decide_action(state))

    def state_summary(self, state):
        return state_summary(state)

    def is_game_over(self, state):
        return state.get("game_state") == "GAMEOVER"

    def on_start(self):
        send_action("RESUME")

    def log_events(self, state):
        score = state.get("score", 0)
        if score != self._last_score:
            print(f"  Score: {score}  (+{score - self._last_score})")
            self._last_score = score

if __name__ == "__main__":
    BreakoutRunner().run()

---

ToolRunner reference

ToolRunner is a base class in omnilink.tool_runner. Subclass it and override the hooks below to build any local tool controller.

Configuration (class attributes)

Attribute	Default	Description
agent_name	"tool-agent"	Agent profile name on OmniLink.
display_name	"Tool"	Human-readable name (used in banners and logs).
base_url	"https://www.omnilink-agents.com"	OmniLink API base URL.
omni_key	os.environ["OMNI_KEY"]	Your Omni Key (reads from OMNI_KEY env var by default).
engine	"g2-engine"	AI engine to use (g1-engine, g2-engine, etc.).
poll_interval	0.0	Seconds between ticks (0 = as fast as possible).
memory_every	60	Save state to agent memory every N seconds.
ask_every	2400	Periodic agent review interval in seconds.
tool_name	"make_move"	Name of the tool the agent calls.
tool_description	"Execute the next action."	Tool description for the agent.
commands	"stop_game, pause_game, resume_game"	Available UI commands.
game_server_url	None	URL of the local game/system server (for pause/resume commands).
tool_callback_url	None	URL the UI calls for tool execution. Auto-set when query_tools is defined.
query_tools	[]	List of query tool definitions: [{"name": "get_score", "description": "...", "parameters": {"type": "object", "properties": {}}}].
default_tools	[]	List of DefaultTool subclass instances with parameters_schema.

Required hooks (must override)

Method	Signature	Description
get_state()	→ dict	Fetch current state from the target system.
execute_action(state)	→ None	Decide and send the next action.
state_summary(state)	→ str	Concise text summary for agent memory.
is_game_over(state)	→ bool	Return True when the task is finished.

Optional hooks

Method	Default	Description
on_start()	No-op	Called after kickoff, before the main loop (e.g. send RESUME).
log_events(state)	No-op	Print noteworthy events each tick.
game_over_message(state)	"GAME OVER"	Text for the game-over banner.
execute_query_tool(name, **kwargs)	Returns error	Handle query tool calls from the UI.
get_system_instruction()	Auto-generated	Override for custom kickoff prompt.
get_review_instruction()	Auto-generated	Override for custom review prompt.
get_profile_settings()	Auto-generated	Override for custom agent profile.

Built-in behaviour

The run() method handles the full lifecycle automatically:

1. Profile setup — creates or updates the agent profile
2. Tool-call kickoff — one API call to trigger the tool
3. Main loop — calls get_state() → is_game_over() → execute_action() → log_events()
4. Memory persistence — saves state_summary() every memory_every seconds
5. UI command polling — checks memory for stop_game / pause_game / resume_game
6. Periodic review — asks the agent to review and decide continue/stop every ask_every seconds
7. Final analysis — saves final state and asks the agent for a summary

---

Built-in examples

The library ships with 11 benchmark examples and a UI-controlled robot demo, each following the same three-file pattern: play_*.py (ToolRunner subclass), *_api.py (HTTP client to the target system), and *_engine.py (local AI logic).

Game benchmarks

Example	Agent Name	AI Strategy	Key Algorithms
Pac-Man	pacman-agent	Survival-first BFS	BFS pathfinding, ghost avoidance, dead-end detection, junction preference
Chess	chess-agent	Minimax search	Alpha-beta pruning, piece-square tables, move ordering
Tetris	tetris-agent	Optimal placement	Pierre Dellacherie evaluation, macro action batching
Breakout	breakout-agent	Ball prediction	Trajectory simulation with brick collision, paddle deflection physics
Pong	pong-agent	Full simulation	Ball trajectory with opponent AI modelling, integer-exact positioning
Space Invaders	space-invaders-agent	Lead-shot targeting	Stateful direction tracking, wall-bounce reflection, threat scoring
Montezuma	montezuma-agent	Safe pathfinding	BFS with enemy avoidance radius, fallback strategies
Go	go-agent	Heuristic evaluation	Capture priority, atari rescue, influence scoring, territory estimation
Asteroids	asteroids-agent	Multi-tool strategic	Engage/evade/analyze modes, trajectory simulation, threat mapping

Robotics benchmarks

Example	Agent Name	Description
Husky	husky-agent	Single robot waypoint navigation with PyBullet physics
Husky Fleet	husky-fleet-agent	5-robot fleet coordination with independent waypoint circuits

UI-controlled demo

Example	Agent Name	Description
Robot Demo	robot-demo	Interactive 2D grid robot — user sends commands and tool queries via the OmniLink UI, demo executes them against a Pygame simulation

Each example includes query tools that the OmniLink UI can call in real time via toolCallbackUrl (see Connecting tools to the UI).

Run any example with:

export OMNI_KEY="olink_YOUR_KEY_HERE"

# Game benchmarks (start the game server first)
python -m omnilink.examples.pacman.play_pacman
python -m omnilink.examples.chess.play_chess
python -m omnilink.examples.tetris.play_tetris
python -m omnilink.examples.asteroids.play_asteroids

# Robotics (start the PyBullet simulator first)
python -m omnilink.examples.husky.play_husky
python -m omnilink.examples.husky_fleet.play_fleet

# UI-controlled robot demo (native tool calling)
python -m omnilink.examples.robot_demo.robot_sim         # Terminal 1: Pygame sim
python -m omnilink.examples.robot_demo.run_tools_demo    # Terminal 2: ToolRunner

Each benchmark requires its corresponding server running locally. See the omnilink-benchmarks/ directory for the game servers and simulators.

---

Core concepts — command engine

The command engine is the second major system in OmniLink. While ToolRunner handles cloud-orchestrated local tools, the command engine handles natural-language command routing — useful for chatbots, controllers, and interactive agents.

Before writing any code, it helps to understand the four building blocks:

Concept	What it does
OmniLinkEngine	Matches incoming text against your templates, extracts variables, and calls the right handler.
Templates	Plain-English patterns that describe the commands your agent understands, e.g. "turn on the [room] lights".
Handlers	Python functions you write. The engine calls them when a template matches.
Bridges	Optional transport adapters. A bridge connects the engine to a network via HTTP so external systems can send commands and receive replies.
AgentMessenger	An object injected into your handler by a bridge. It lets your handler send progress updates, ask the operator a question, and acknowledge receipt.

You can use the engine alone (direct Python calls) or attach a bridge for network connectivity. Handlers always look the same regardless of which bridge is in use.

---

Quick start — engine only

The simplest possible setup: create an engine, register a handler, call it.

from omnilink import OmniLinkEngine

engine = OmniLinkEngine([
    "hello",
    "echo [message:any]",
])

def handle_hello(event):
    return {"message": "Hello, world!"}

def handle_echo(event):
    text = event["vars"]["message"].replace("_", " ")
    return {"echo": text}

engine.on_template("hello", handle_hello)
engine.on_template("echo [message:any]", handle_echo)

result = engine.handle("hello")
print(result["result"])          # {"message": "Hello, world!"}

result = engine.handle("echo good morning")
print(result["result"])          # {"echo": "good morning"}

engine.handle() always returns a dict that includes:

Key	Meaning
ok	True if a handler ran without error, False otherwise.
template	The template that matched (or None).
vars	Dict of extracted variables.
result	Whatever your handler returned.
errors	List of type-conversion errors, if any.

---

Templates and types

Defining templates

Templates are plain strings. Spaces and underscores are interchangeable — the engine normalises both before matching.

templates = [
    "status",                              # no variables
    "launch [vehicle]",                    # one variable (any word)
    "set speed to [speed:float]",          # typed variable
    "move [color] [piece] to [square]",    # multiple variables
    "say [message:any]",                   # greedy — matches multiple words
]

Variable syntax

Syntax	Matches
[name]	Any single word (letters, digits, underscores, hyphens).
[name:int]	Integer — converted automatically.
[name:float]	Float — converted automatically.
[name:any]	One or more words (greedy).
[name:alpha]	Letters only.
[name:uuid]	UUID string.
[name:/regex/]	Custom regular expression.

Registering custom types

from omnilink import OmniLinkEngine, TypeRegistry

types = TypeRegistry()
types.register(
    "room",
    r"(?:living_room|kitchen|bedroom|office)",   # regex pattern
    lambda raw: raw.replace("_", " "),           # optional converter
)

engine = OmniLinkEngine(
    ["turn on the [room:room] lights"],
    types=types,
)

Loading templates from a file

Keep templates in a text file (one per line, # for comments):

# commands.txt
turn on the [room:room] lights
turn off the [room:room] lights
set thermostat to [temp:float] degrees
lock the [door:door]

from omnilink import load_patterns_from_file

templates = load_patterns_from_file("commands.txt")
engine = OmniLinkEngine(templates, types=types)

---

Writing handlers

The event dictionary

Every handler receives a single event dict:

def my_handler(event: dict) -> dict:
    event["command"]    # raw input text, e.g. "launch falcon9"
    event["template"]   # matched template, e.g. "launch_[vehicle]"
    event["vars"]       # extracted vars, e.g. {"vehicle": "falcon9"}
    event["meta"]       # metadata passed by the caller or bridge
    event["messenger"]  # AgentMessenger (only when a bridge is running)
    event["timestamp"]  # Unix timestamp

Registering handlers

# Match a specific template by name
engine.on_template("launch [vehicle]", handle_launch)

# Match with a custom predicate
engine.on(lambda e: e["vars"].get("speed", 0) > 100, handle_high_speed)

# Fallback — runs when nothing else matches
engine.on(lambda e: e["template"] is None, handle_unknown)

Middleware

Run a function before or after every command:

def log_command(event):
    print(f"[{event['timestamp']}] {event['command']}")

engine.before(log_command)

Returning results

Return any JSON-serialisable dict. The value lands in result["result"]. Return None to indicate a no-op.

---

Bidirectional messaging

When your engine is connected to a bridge, the event["messenger"] object lets your handler communicate back to the operator in real time.

Acknowledge receipt

messenger = event["messenger"]
messenger.acknowledge("pending", message="Starting sequence...")

Send progress updates

from omnilink import AgentFeedback

messenger.send_feedback(AgentFeedback(
    message="Pressurizing tanks",
    kind="info",       # "info" | "success" | "warning" | "error"
    progress=0.4,      # 0.0 – 1.0
    ok=True,
))

Ask the operator a question

Your handler blocks at pending.wait() until the operator replies. Other commands queued behind it continue to be processed normally.

from omnilink import AgentQuestion

question = AgentQuestion(
    prompt="Confirm liftoff?",
    choices=["yes", "no"],
    data={"vehicle": event["vars"]["vehicle"]},
)
pending = messenger.ask_question(question)

try:
    reply = pending.wait(timeout=30, cancel_on_timeout=True)
except TimeoutError:
    return {"status": "timed out"}

if reply.get("answer") == "yes":
    return {"status": "launched"}
return {"status": "aborted"}

---

Connecting over HTTP

OmniLinkHTTPBridge turns your engine into a local HTTP server. Any client that can make HTTP requests — a browser, curl, another Python script — can send commands and receive responses.

Start the server

from omnilink import OmniLinkEngine, OmniLinkHTTPBridge

engine = OmniLinkEngine(["launch [vehicle]"])
engine.on_template("launch [vehicle]", handle_launch)

bridge = OmniLinkHTTPBridge(engine, host="0.0.0.0", port=8080)
bridge.loop_forever()   # blocks; use bridge.start() for background mode

Sending a command

curl -s -X POST http://localhost:8080/command \
  -H "Content-Type: application/json" \
  -d '{"command": "launch falcon9"}'

Response:

{"status": "ok", "command": "launch falcon9"}

The command is processed asynchronously in a background thread.

Polling for feedback

curl http://localhost:8080/feedback

Returns the latest feedback from the most recent command.

Endpoint layout

Method	Path	Description
POST	/command	Submit a command (JSON body with command field).
GET	/feedback	Latest feedback from the last processed command.
GET	/context	Latest context data / state snapshot.
POST	/inline-code	Submit inline code snippets.

Publishing state snapshots

bridge.publish_context("System is ready")
bridge.publish_state_snapshot(history_limit=10)

Configuration

Environment variable	Default	Description
HTTP_BRIDGE_HOST	0.0.0.0	Bind address.
HTTP_BRIDGE_PORT	8080	Bind port.

---

REST API client

OmniLinkClient is a full Python client for the OmniLink platform. It covers chat, agent profiles, memory, speech-to-text, text-to-speech, and translation.

from omnilink.client import OmniLinkClient

client = OmniLinkClient(omni_key="olink_...")

# Chat
reply = client.chat("What is 2 + 2?", agent_name="math-tutor")
print(reply["text"])

# Agent profiles
profile = client.create_profile("my-agent", settings={
    "agentName": "my-agent",
    "mainTask": "You are a helpful assistant.",
    "allowToolUse": True,
    "availableTools": "make_move",
    "availableToolDetails": [{"name": "make_move", "description": "Run the AI."}],
    "availableCommands": "stop_game, pause_game, resume_game",
})
profiles = client.list_profiles()
client.update_profile(profile["id"], name="my-agent", settings={...})
client.delete_profile(profile["id"])

# Save and retrieve conversation memory
client.set_memory("my-agent", [
    {"role": "user",  "parts": [{"text": "Hello"}]},
    {"role": "model", "parts": [{"text": "Hi!"}]},
])
memory = client.get_memory("my-agent")
client.clear_memory("my-agent")

# Speech-to-text
with open("audio.webm", "rb") as f:
    result = client.transcribe(f.read(), mime_type="audio/webm")
print(result["text"])

# Text-to-speech
audio = client.synthesize_to_bytes("Hello from OmniLink!")
with open("out.mp3", "wb") as f:
    f.write(audio)

# Translation
result = client.translate("Bonjour le monde", target_language="English")
print(result["translatedText"])

OmniLinkClient is independent of OmniLinkEngine — use it to communicate with a remote OmniLink deployment from any Python script. There is also a lightweight OmniLinkChatClient for chat-only use cases.

---

Connecting tools to the UI

The OmniLink UI executes tools via native structured tool calling. Each AI engine sends tool definitions in the provider's native API format (OpenAI function calling, Anthropic tool_use, Gemini functionDeclarations). The provider responds with structured tool call objects — no text parsing needed.

How it works

User: "What's the score?"
  → Engine sends tools via native API (e.g. OpenAI tools[].function)
  → Provider responds with structured tool_call: {name: "get_score", arguments: {}}
  → UI POSTs to toolCallbackUrl: {"tool": "get_score"}
  → Your server returns: {"result": {"score": 1500, "lives": 3}}
  → AI summarises: "The score is 1500 with 3 lives remaining."
  → User sees the summary

Adding query tools to a ToolRunner

1. Define query_tools with JSON Schema parameters:

class MyRunner(ToolRunner):
    query_tools = [
        {
            "name": "get_score",
            "description": "Returns the current score.",
            "parameters": {"type": "object", "properties": {}},
        },
        {
            "name": "get_position",
            "description": "Returns player position.",
            "parameters": {"type": "object", "properties": {}},
        },
    ]

Important: Every tool must include a parameters field. Use {"type": "object", "properties": {}} for tools with no arguments. This is required for native tool calling — providers like OpenAI reject function definitions without a parameters field.

2. Override execute_query_tool():

    def execute_query_tool(self, tool_name, **kwargs):
        state = self.get_state()
        if tool_name == "get_score":
            return {"score": state.get("score", 0)}
        if tool_name == "get_position":
            return {"x": state.get("x"), "y": state.get("y")}
        return {"error": f"Unknown tool: {tool_name}"}

3. Run — the ToolRunner automatically:
   • Starts a lightweight HTTP server on a random port
   • Sets toolCallbackUrl in the agent profile
   • The UI can now call your tools directly

Manual toolCallbackUrl

If your target system already has a /tool endpoint with CORS headers, set tool_callback_url directly instead of using query tools:

class MyRunner(ToolRunner):
    tool_callback_url = "http://127.0.0.1:5050/tool"

The endpoint must accept POST requests with {"tool": "<name>", ...args} and return {"result": {...}}. CORS headers (Access-Control-Allow-Origin: *) are required for browser access.

---

Environment variable reference

Variable	Default	Description
OMNI_KEY	—	Your Omni Key. Used by ToolRunner and as a fallback for OmniLinkClient.
HTTP_BRIDGE_HOST	0.0.0.0	HTTP bridge bind address.
HTTP_BRIDGE_PORT	8080	HTTP bridge bind port.

---

Running the examples

All examples live under src/omnilink/examples/. The library includes:

Standalone scripts

# Minimal engine demo — no bridge, no key needed
python -m omnilink.examples.hello_world

# Chat client quickstart (requires OMNI_KEY)
OMNI_KEY=olink_... python -m omnilink.examples.chat_quickstart

# Full REST API client walkthrough
OMNI_KEY=olink_... python -m omnilink.examples.client_demo

# Tool-calling with arithmetic operations
OMNI_KEY=olink_... python -m omnilink.examples.arithmetic_tools

ToolRunner benchmark games

Each game requires its corresponding game server from omnilink-benchmarks/:

# Start the game server first, then run the agent:
OMNI_KEY=olink_... python -m omnilink.examples.pacman.play_pacman
OMNI_KEY=olink_... python -m omnilink.examples.chess.play_chess
OMNI_KEY=olink_... python -m omnilink.examples.tetris.play_tetris
OMNI_KEY=olink_... python -m omnilink.examples.breakout.play_breakout
OMNI_KEY=olink_... python -m omnilink.examples.pong.play_pong
OMNI_KEY=olink_... python -m omnilink.examples.space_invaders.play_space_invaders
OMNI_KEY=olink_... python -m omnilink.examples.montezuma.play_montezuma
OMNI_KEY=olink_... python -m omnilink.examples.go.play_go
OMNI_KEY=olink_... python -m omnilink.examples.asteroids.play_asteroids

Robotics examples

# Husky (start husky_drive.py from omnilink-benchmarks first)
OMNI_KEY=olink_... python -m omnilink.examples.husky.play_husky

# Husky Fleet (start husky_fleet.py from omnilink-benchmarks first)
OMNI_KEY=olink_... python -m omnilink.examples.husky_fleet.play_fleet

UI-controlled robot demo

# Terminal 1: Start the Pygame simulation
python -m omnilink.examples.robot_demo.robot_sim

# Terminal 2: Start the UI bridge
OMNI_KEY=olink_... python -m omnilink.examples.robot_demo.run_demo

# Open https://www.omnilink-agents.com, select "robot-demo", and type commands.

---

Running the tests

pytest omnilink-lib/tests