Skip to content

Tutorial: Your First Plugin

This tutorial shows the complete lifecycle of a Stepyard plugin - from scaffolding to running in a flow. You will build a plugin that sends a message to a Telegram chat.

What you'll learn:

  • Create a plugin with stepyard plugin init
  • Write a @node function
  • Validate inputs with type hints
  • Install the plugin into a project
  • Write a unit test with stepyard.sdk.testing
  • Use the node in a flow

Estimated time: 15 minutes.

1. Scaffold the plugin

stepyard plugin init stepyard-plugin-telegram ./stepyard-plugin-telegram
cd stepyard-plugin-telegram

The scaffold creates:

stepyard-plugin-telegram/
├── pyproject.toml
├── README.md
├── src/
│   └── stepyard_plugin_telegram/
│       ├── __init__.py
│       └── nodes.py
└── tests/
    └── test_nodes.py

2. Write the node

Open src/stepyard_plugin_telegram/nodes.py and replace it with:

src/stepyard_plugin_telegram/nodes.py
import httpx
from stepyard.sdk import node, NodeResult


@node(name="telegram.send_message")
def send_message(
    token: str,          # (1)
    chat_id: str,
    text: str,
    parse_mode: str = "HTML",
    disable_notification: bool = False,
) -> NodeResult:
    """Send a message to a Telegram chat via the Bot API.

    Args:
        token: Telegram Bot API token (keep in env, not in YAML).
        chat_id: Numeric chat ID or @username of the channel.
        text: Message body. Supports HTML or Markdown depending on parse_mode.
        parse_mode: "HTML" or "MarkdownV2". Default: "HTML".
        disable_notification: Send silently (no sound/vibration).
    """
    url = f"https://api.telegram.org/bot{token}/sendMessage"

    response = httpx.post(
        url,
        json={
            "chat_id": chat_id,
            "text": text,
            "parse_mode": parse_mode,
            "disable_notification": disable_notification,
        },
        timeout=10,
    )

    data = response.json()

    if not data.get("ok"):
        raise RuntimeError(f"Telegram API error: {data.get('description', 'unknown')}")

    return NodeResult(
        status="success",
        output={
            "message_id": data["result"]["message_id"],
            "chat_id": data["result"]["chat"]["id"],
        },
    )
  1. Stepyard reads these type hints and builds a Pydantic validation model automatically. If a flow passes chat_id: 123 (int) where a str is expected, validation fails before the function is called.

3. Declare the entry point

The pyproject.toml generated by plugin init already has the entry-point group. Make sure it lists your node file:

pyproject.toml
[project.entry-points."stepyard.plugins"]
telegram = "stepyard_plugin_telegram.nodes"

4. Write a unit test

Open tests/test_nodes.py:

tests/test_nodes.py
import pytest
from unittest.mock import patch, MagicMock
from stepyard.sdk.testing import run_node

from stepyard_plugin_telegram.nodes import send_message


def test_send_message_success():
    mock_response = MagicMock()
    mock_response.json.return_value = {
        "ok": True,
        "result": {"message_id": 42, "chat": {"id": 123456}},
    }

    with patch("httpx.post", return_value=mock_response):
        result = run_node(                # (1)
            send_message,
            {"token": "bot-token", "chat_id": "123456", "text": "Hello from tests!"},
        )

    assert result.output["message_id"] == 42
    assert result.output["chat_id"] == 123456


def test_send_message_api_error():
    mock_response = MagicMock()
    mock_response.json.return_value = {
        "ok": False,
        "description": "chat not found",
    }

    with patch("httpx.post", return_value=mock_response):
        with pytest.raises(RuntimeError, match="Telegram API error"):
            run_node(send_message, {"token": "t", "chat_id": "bad", "text": "hi"})


def test_requires_token():
    """Passing nothing should raise a validation error before any network call."""
    with pytest.raises(Exception, match="token"):
        run_node(send_message, {"chat_id": "123", "text": "hi"})  # token missing
  1. run_node (synchronous) and invoke_node (async) run your function through the same Pydantic validation and NodeContext injection that the real engine uses - no mocking of Stepyard internals required. Access outputs via result.output.

Run the tests:

pytest tests/ -v

5. Install the plugin into your project

Go back to your Stepyard project and install the plugin:

cd ../ci-demo
stepyard plugin add ../stepyard-plugin-telegram

Stepyard installs the package into .stepyard/env (an isolated virtualenv) and records it in stepyard.lock.

Verify the node is visible (tools list shows nodes; plugin list shows packages):

stepyard tools list

Node                     Source
─────────────────────────────────────────────────
shell.run                builtin
http.request             builtin
llm.generate             builtin
telegram.send_message    stepyard-plugin-telegram

6. Use it in a flow

flows/ci.yaml
  # ... previous steps ...

  - id: notify_success
    if: ${{ steps.test.output.code == 0 }}
    uses: telegram.send_message
    with:
      token: ${{ env.TELEGRAM_BOT_TOKEN }}    # (1)
      chat_id: ${{ env.TELEGRAM_CHAT_ID }}
      text: |
        ✅ <b>CI passed</b>

        Branch: ${{ env.BRANCH }}
        Tests: ${{ steps.test.output.stdout }}
  1. Never hardcode tokens in YAML. Use ${{ env.VAR }} and pass real values through environment variables or a .env file.

Run:

TELEGRAM_BOT_TOKEN=... TELEGRAM_CHAT_ID=... stepyard run ci

7. Iterate on the plugin

Installs into .stepyard/env are regular (non-editable) packages, so after you change the plugin's source, re-run plugin add to pick up the new code and re-discover its capabilities:

stepyard plugin add ../stepyard-plugin-telegram

Confirm the node is registered with stepyard tools list.

What's next?