Cortex Writes Its Own Tools#

In the Cortex tutorials so far, the planner has been given a set of tools and has chosen among them, which works as long as the recipe has a tool for everything a task might need. It rarely does. Sooner or later a task needs something nobody wired in, such as a way to send an email or a sensor reading that no component publishes, and until now the only answer was to stop the robot, write the missing component or action, and start again. This tutorial turns on enable_scratch_functions, which lets the planner write the missing piece itself as a Python function and put it to use immediately.

To see it at work, we build a small recipe with a camera, object detection and a voice, and then give Cortex two standing instructions that it cannot carry out with what the recipe provides:

  • “Whenever a person is detected, email me about it.” There is no email action anywhere in the recipe.

  • “Whenever the CPU temperature goes above 80 degrees, say so out loud.” No topic publishes the CPU temperature.

The planner handles the first one by writing an action and installing an event on the detections topic that runs it, and the second by writing a condition and installing an event that polls that condition and calls the voice. Both therefore combine the standing-instruction tools with the new function-writing tools, each of which the Cortex page explains on its own, under Standing instructions and Functions the planner writes.

Warning

A function the planner writes runs in the launcher process, with that process’s privileges, and it can do anything Python can do. That is why the feature is off by default, and why it should only be turned on with a planning model and on a deployment that you trust to that degree.

Step 1: The capability components#

The recipe has two ordinary components: a Vision component that publishes detections from the camera, and a TextToSpeech component that speaks whatever arrives on its input topic and also offers a say action, which is what the planner will end up calling. Both use models served by RoboML, as in Cortex: The Agentic Harness.

from agents.clients import RoboMLRESPClient
from agents.components import TextToSpeech, Vision
from agents.config import TextToSpeechConfig, VisionConfig
from agents.models import TransformersTTS, VisionModel
from agents.ros import Topic

detection_client = RoboMLRESPClient(
    VisionModel(name="rtdetr", checkpoint="PekingU/rtdetr_r50vd_coco_o365")
)
tts_client = RoboMLRESPClient(TransformersTTS(name="speecht5"))

image_in = Topic(name="/image_raw", msg_type="Image")
detections_out = Topic(name="detections", msg_type="Detections")

vision = Vision(
    inputs=[image_in],
    outputs=[detections_out],
    model_client=detection_client,
    config=VisionConfig(threshold=0.5),
    trigger=0.5,
    component_name="vision",
)

text_in = Topic(name="text_in", msg_type="String")

tts = TextToSpeech(
    inputs=[text_in],
    model_client=tts_client,
    config=TextToSpeechConfig(play_on_device=True),
    trigger=text_in,
    component_name="tts",
)

Step 2: A planner that may write code#

Three settings on CortexConfig make the difference. enable_events gives the planner the standing-instruction tools, which both of our instructions need, and enable_scratch_functions lets it write functions. The third, scratch_notes, tells the planner what the environment holds, and it matters more than it looks, because the planner cannot see your environment variables or your filesystem. This note is where it learns that an SMTP account is available and where the CPU temperature can be read, and without it the code it writes would be guesswork.

from agents.clients import OllamaClient
from agents.components import Cortex
from agents.config import CortexConfig
from agents.models import OllamaModel

planner_client = OllamaClient(
    OllamaModel(name="qwen", checkpoint="qwen3.5:latest", think=False),
    inference_timeout=120,
)

cortex = Cortex(
    model_client=planner_client,
    config=CortexConfig(
        enable_events=True,
        enable_scratch_functions=True,
        scratch_notes=(
            "The process environment holds SMTP_HOST, SMTP_PORT, SMTP_USER and "
            "SMTP_PASSWORD, an SMTP server and account for sending mail, and "
            "ALERT_EMAIL, the operator's address. The CPU temperature in "
            "millidegrees Celsius can be read from "
            "/sys/class/thermal/thermal_zone0/temp."
        ),
        max_new_tokens=2000,
    ),
    component_name="cortex",
)

Two settings on the model side matter just as much. Since a written function’s source has to arrive in a single reply, max_new_tokens needs a generous budget, and 2000 is a sensible floor, with more doing no harm. And because a thinking model that is left to think spends that same budget before it starts answering, and plans slowly besides, think=False on the OllamaModel turns thinking off for this job.

Note

scratch_notes only names the environment variables; it never holds their values. The credentials stay in the environment of the process that runs the recipe, and a written function reads them from there with os.environ.

Step 3: Launch#

from agents.ros import Launcher

launcher = Launcher()
launcher.add_pkg(
    components=[vision, tts, cortex],
    package_name="automatika_embodied_agents",
    multiprocessing=True,
)
launcher.bringup()

The recipe expects the SMTP settings and the operator’s address in the environment it is started from, so export them first:

export SMTP_HOST=mail.example.org SMTP_PORT=587 SMTP_USER=[email protected]
export SMTP_PASSWORD=... ALERT_EMAIL=[email protected]
python3 recipe.py

The first instruction: email when a person appears#

Send the standing instruction as a goal, from a terminal or from the web interface:

ros2 action send_goal /cortex_input_command \
    automatika_embodied_agents/action/VisionLanguageAction \
    "{task: 'Whenever a person is detected, email me about it.'}"

The planner looks through its tools and finds the detections topic, the tts-say action and the event tools, but nothing that sends mail, so it writes an action. Exactly what it writes depends on the model, but the contract it has to keep does not: the function has typed parameters and a docstring, returns (True, message) or (False, why) and never raises, imports what it needs, and reads the environment through os.environ. It will look something like this:

def email_operator(subject: str, body: str) -> tuple[bool, str]:
    """Email the operator at ALERT_EMAIL through the SMTP account in the environment.

    subject: The subject line
    body: The text of the message
    """
    import os
    import smtplib
    from email.message import EmailMessage

    message = EmailMessage()
    message["From"] = os.environ["SMTP_USER"]
    message["To"] = os.environ["ALERT_EMAIL"]
    message["Subject"] = subject
    message.set_content(body)
    try:
        with smtplib.SMTP(os.environ["SMTP_HOST"], int(os.environ["SMTP_PORT"])) as smtp:
            smtp.starttls()
            smtp.login(os.environ["SMTP_USER"], os.environ["SMTP_PASSWORD"])
            smtp.send_message(message)
    except Exception as e:
        return False, f"Could not send the email: {e}"
    return True, f"Emailed {message['To']}"

Cortex checks the source as soon as it arrives, and if it does not parse, does not keep the signature, has no docstring, or uses a name it did not import, the planner is told why and tries again. Once it passes, the log shows the source under Written action 'email_operator', and the planner learns that its function is now the tool scratch-email_operator, with the docstring as the tool’s description and the subject: ... and body: ... lines as the descriptions of its parameters.

The planner then installs the event, with a condition on the detections topic and the new tool as the action, and since the instruction says whenever, the event is kept rather than fired once:

{
  "event_id": "person_detected_email",
  "conditions": [{"topic": "detections", "field": "labels", "operator": "contains", "value": "person"}],
  "actions": [{"tool": "scratch-email_operator", "arguments": {"subject": "Person detected", "body": "The camera has detected a person."}}],
  "once": false
}

From then on the event lives in the recipe’s monitor and fires on its own each time a person shows up in the detections, whether or not a task is running, and like every standing instruction it outlives the task that installed it.

The second instruction: the CPU temperature#

ros2 action send_goal /cortex_input_command \
    automatika_embodied_agents/action/VisionLanguageAction \
    "{task: 'Whenever the CPU temperature goes above 80 degrees, say so out loud.'}"

This time the action exists, tts-say, but there is nothing to install an event on, because no topic carries the temperature. So the planner writes a condition instead: a function with no parameters that returns a bool, is quick enough to be polled, and returns False rather than raising when it cannot tell:

def cpu_too_hot() -> bool:
    """Whether the CPU temperature is above 80 degrees Celsius."""
    try:
        with open("/sys/class/thermal/thermal_zone0/temp") as f:
            return int(f.read()) > 80000
    except OSError:
        return False

Since a written condition is not a topic, it is given to add_event as a plugin_condition, just like a condition that a plugin provides, such as a low battery, together with the rate at which it should be polled:

{
  "event_id": "cpu_too_hot",
  "plugin_condition": {"name": "scratch-cpu_too_hot", "arguments": {"check_rate": 1.0}},
  "actions": [{"tool": "tts-say", "arguments": {"text": "The CPU is running hot."}}],
  "once": false
}

The monitor polls the function once a second and fires the event when its answer turns true, which is the same mechanism as a condition on a Python object in Internal State Events, only with the planner writing the function. A condition that raises, returns something other than a bool, or takes longer than its polling period counts as not met, and the log says so once.

Looking at what was written#

The planner can list, read and remove its functions, and you can make it do so simply by asking: “list the functions you have written” makes it call list_functions, and the answer names each function with its kind and signature. Writing a name again replaces the function, though an event that was installed with the earlier version keeps that version until the event is removed and added again, and remove_function forgets a function without touching the events that use it.

A written action can also call the recipe’s existing actions through run_action("<tool name>", **arguments), which returns (success, message) like any other action, so a written function can build on what is already there, for instance by saying a message out loud before sending it. The one exception is goals to action servers, which a written function cannot send; the planner is told to plan such a goal as a step instead.

Complete code#

Cortex writing its own actions and event conditions#
 1from agents.clients import OllamaClient, RoboMLRESPClient
 2from agents.components import Cortex, TextToSpeech, Vision
 3from agents.config import CortexConfig, TextToSpeechConfig, VisionConfig
 4from agents.models import OllamaModel, TransformersTTS, VisionModel
 5from agents.ros import Launcher, Topic
 6
 7# --- Model clients ---
 8# A thinking model left to think plans slowly and can use up its token budget
 9# before it answers
10planner_client = OllamaClient(
11    OllamaModel(name="qwen", checkpoint="qwen3.5:latest", think=False),
12    inference_timeout=120,
13)
14detection_client = RoboMLRESPClient(
15    VisionModel(name="rtdetr", checkpoint="PekingU/rtdetr_r50vd_coco_o365")
16)
17tts_client = RoboMLRESPClient(TransformersTTS(name="speecht5"))
18
19# --- Vision: detections the planner can install events on ---
20image_in = Topic(name="/image_raw", msg_type="Image")
21detections_out = Topic(name="detections", msg_type="Detections")
22
23vision = Vision(
24    inputs=[image_in],
25    outputs=[detections_out],
26    model_client=detection_client,
27    config=VisionConfig(threshold=0.5),
28    trigger=0.5,
29    component_name="vision",
30)
31
32# --- TextToSpeech: its `say` action is a tool for the planner ---
33text_in = Topic(name="text_in", msg_type="String")
34
35tts = TextToSpeech(
36    inputs=[text_in],
37    model_client=tts_client,
38    config=TextToSpeechConfig(play_on_device=True),
39    trigger=text_in,
40    component_name="tts",
41)
42
43# --- Cortex: may write functions, and knows what the environment offers ---
44cortex = Cortex(
45    model_client=planner_client,
46    config=CortexConfig(
47        enable_events=True,  # standing instructions need runtime events
48        enable_scratch_functions=True,  # read the warning above first
49        scratch_notes=(
50            "The process environment holds SMTP_HOST, SMTP_PORT, SMTP_USER and "
51            "SMTP_PASSWORD, an SMTP server and account for sending mail, and "
52            "ALERT_EMAIL, the operator's address. The CPU temperature in "
53            "millidegrees Celsius can be read from "
54            "/sys/class/thermal/thermal_zone0/temp."
55        ),
56        max_new_tokens=2000,  # a written function's source travels in one reply
57    ),
58    component_name="cortex",
59)
60
61launcher = Launcher()
62launcher.add_pkg(
63    components=[vision, tts, cortex],
64    package_name="automatika_embodied_agents",
65    multiprocessing=True,
66)
67launcher.bringup()

Where next#

  • Cortex – the reference for standing instructions and for the functions the planner writes, including the full contract a function has to keep.

  • Events & Actions – what an event can watch and what it can run, which is what the planner’s add_event calls are built from.

  • Internal State Events – a polled condition on a Python object, written by hand; a written condition is the same thing written by the planner.

  • Event-Driven Cognition – the recipes where events wake expensive models only when needed, the pattern the first instruction reproduces at run time.

Tip

Promote this recipe to production. While you are shaping it, run the script directly with python recipe.py. Once it is solid, drop it at ~/emos/recipes/<name>/recipe.py and start it with emos run <name>, or from the dashboard. Either way every run is logged under ~/emos/logs, and an operator gets a card to launch it from a browser. Running Recipes covers the two ways of running a recipe and what differs per install mode.