External Services & APIs

Some skills reach beyond the robot’s body — sending emails, calling APIs, retrieving information. These are ordinary code-defined skills that happen to talk to the network: they implement explicit protocols and handle authentication, errors, and connectivity.

What to keep in mind

Talking to an external service is a more deterministic domain than acting in the physical world, but it comes with its own constraints:

• Protocol-based: follow defined APIs and standards

• Atomic: many operations cannot be cancelled once started

• Reliable: once working, behavior is consistent

• Network-dependent: must handle connectivity issues

Built-in examples

SendEmail

Sends email notifications, typically for alerts or status updates.

class SendEmail(Skill):
    """Send an emergency email notification. Provide a subject and message,
    and optionally a list of recipients — otherwise the default list is used.
    Use when a potential emergency is detected and assistance might be
    required."""

    def execute(self, subject: str, message: str, recipients: list[str] | None = None):
        # Send via SMTP
        return f"Sent to {len(recipients or DEFAULTS)} recipients"

Parameters:

SendPictureViaEmail

Sends an email with the robot’s current camera view attached.

class SendPictureViaEmail(Skill):
    """Send an email with a photo of what the robot can currently see."""

    image: MainImage        # guaranteed inside execute() — no None check needed

    def execute(self, subject: str, message: str, recipient: str | None = None):
        attachment = self.image.jpeg     # raw JPEG bytes
        ...

This skill shows the value of a declared feed: image: MainImage makes the runtime wait for a real frame and fail the run up front if none arrives, so execute() never has to handle a missing picture.

RetrieveEmails

Fetches recent emails from the configured account.

class RetrieveEmails(Skill):
    """Retrieve recent emails from the configured account. Provide how many to
    fetch (default 5). Returns subjects and content."""

    def execute(self, count: int = 5):
        ...

Want to build your own version against your own account? There’s a complete implementation in the worked example below.

Building a service skill

Template

import os

from innate import Skill


class MyServiceSkill(Skill):
    """One tight paragraph telling the agent when to call this and what it
    needs — this docstring is what the agent reads."""

    def execute(self, param: str):
        api_key = os.environ.get("SERVICE_API_KEY")
        if not api_key:
            self.fail("SERVICE_API_KEY is not configured on this robot")

        try:
            result = call_service(param, api_key, timeout=10)
        except TimeoutError:
            self.fail("The service timed out")

        return f"Success: {result}"

Read credentials inside execute(), not in an __init__: skill instances are built per run, and a constructor that raises takes the whole skill off the roster instead of failing one run with a message the agent can read.

Worked example: RetrieveEmails

A complete, runnable custom skill that fetches your latest Gmail messages over IMAP. Save it as ~/innate-os/workspace/custom_skills/retrieve_emails.py on the robot. The class name is the skill name, so this is local/retrieve_emails.

import imaplib

from innate import Skill

IMAP_SERVER = "imap.gmail.com"
ADDRESS = "[e-mail]"
# Use a Gmail App Password (https://myaccount.google.com/apppasswords),
# not your main account password.
PASSWORD = "your_app_password"


class RetrieveEmails(Skill):
    """Retrieve recent emails from the configured account. Provide how many to
    fetch (default 5, max 20). Returns subjects and content."""

    def execute(self, count: int = 5):
        count = min(max(1, count), 20)
        try:
            mail = imaplib.IMAP4_SSL(IMAP_SERVER, 993)
        except Exception as e:
            self.fail(f"Could not reach {IMAP_SERVER}: {e}")

        try:
            mail.login(ADDRESS, PASSWORD)
            # ... fetch and process emails ...
            self.feedback("Email 1: Subject, From, Content...")
        except Exception as e:
            self.fail(f"Failed to retrieve emails: {e}")
        finally:
            mail.logout()

        return f"Retrieved {count} emails with subjects and content"

Best practices

Authentication

• Store credentials in environment variables or a secret manager

• Never hardcode passwords or API keys

• Fail the run with a clear message when a credential is missing, so the agent can say so

Error handling

Call self.fail() rather than returning an error — it raises, so there is no path where a failure silently reads as success:

def execute(self, query: str):
    try:
        response = self.client.call(query, timeout=10)
    except RateLimitError:
        self.fail("Rate limit exceeded")
    except NetworkError:
        self.fail("Network unavailable")
    return f"Result: {response}"

An uncaught exception also fails the run with its message, so you only catch what you can say something useful about.

Timeouts

• Always set explicit timeouts on network calls

• A blocking network call is not a cancel point — a Stop can’t interrupt it, so keep the timeout short enough that the robot stays responsive

Idempotency

• Design operations to be safely retryable where possible

• Consider partial failure scenarios

Reading robot state

A service skill often wants to send something the robot can see or know. Declare it the same way as any other feed:

from innate import MainImage, Pose, Skill


class ReportPosition(Skill):
    """Email a photo and the robot's current map position."""

    image: MainImage
    pose: Pose

    def execute(self):
        body = f"I'm at ({self.pose.x:.1f}, {self.pose.y:.1f}) on the map."
        send_email(body, attachment=self.image.jpeg)
        return "Report sent"

Useful feeds for a service skill:

See Robot state for the full list.

Cancellation

Many service operations are atomic and can’t be meaningfully interrupted. That’s fine — you don’t write anything for it. The framework latches the cancel, and the run reports CANCELLED once your call returns; a network request in flight simply finishes first.

If a request is long enough that this matters, break the work into steps and call self.check_cancelled() between them so a Stop lands promptly.