Nathan Standards

Nathan Development Standards

Standards and patterns for developing within the Nathan project - an n8n-Jira agent automation system.

When to Use

Invoke this skill when:

• Creating or modifying n8n workflow JSON files
• Writing Python code for the Nathan helpers or templating modules
• Designing webhook command contracts
• Building workflow registry configurations
• Implementing spec-driven features via agent-os

Project Architecture

Nathan follows a layered architecture:

External Service (Jira) <-- n8n Workflows <-- Python Agent Service
                             (credentials)      (webhook calls)

Core Principle: n8n owns all external credentials. Python services call n8n webhooks with shared secret authentication.

n8n Workflow Standards

For detailed workflow patterns, load references/n8n-workflow-patterns.md.

Standard Workflow Structure

Every webhook workflow must follow this pattern:

Webhook --> Validate Secret --> Operation --> Respond to Webhook
               |                   |              |
               v                   v              v
           Unauthorized       Error Response   Success Response
           Response (401)     (500)            (200)

Required Node Pattern

{
  "id": "validate-secret",
  "name": "Validate Secret",
  "type": "n8n-nodes-base.if",
  "typeVersion": 2,
  "parameters": {
    "conditions": {
      "conditions": [{
        "leftValue": "={{ $json.headers['x-n8n-secret'] }}",
        "rightValue": "={{ $env.N8N_WEBHOOK_SECRET }}",
        "operator": { "type": "string", "operation": "equals" }
      }]
    }
  }
}

Response Format

All responses must follow this shape:

{ "success": true, "data": {...}, "status_code": 200, "error": null }
{ "success": false, "data": {}, "status_code": 500, "error": "message" }

JQL Expression Escaping

In n8n expressions within JSON, escape properly:

Python Standards

For detailed patterns, load references/python-patterns.md.

Module Structure

nathan/
  helpers/           # Shared utilities (workflow registry, etc.)
  workflows/         # n8n workflow JSON + registry.yaml per category
  templating/        # YAML-to-JSON template engine
  scripts/           # Standalone runnable scripts

Code Style

# Required imports pattern
from __future__ import annotations
from typing import Any
from pathlib import Path
import logging

logger = logging.getLogger(__name__)

# Type hints required, use T | None not Optional[T]
async def trigger_workflow(url: str, params: dict[str, Any]) -> dict[str, Any]:
    ...

Registry Pattern

# registry.yaml
version: "1.0.0"
description: "Registry description"

commands:
  command_name:
    endpoint: /webhook/endpoint-path
    method: POST
    required_params:
      - param1
    optional_params:
      - param2
    description: What this command does
    example:
      param1: "value"

Spec-Driven Development

Use agent-os commands for feature development:

1. /shape-spec - Initialize and shape specification 2. /write-spec - Write detailed spec document 3. /create-tasks - Generate task list from spec 4. /orchestrate-tasks - Delegate to subagents

Specs live in agent-os/specs/[spec-name]/ with:

• spec.md - Feature specification
• tasks.md - Implementation tasks with checkboxes
• orchestration.yml - Subagent delegation config

Quick Reference

Common Commands

uv sync                              # Install dependencies
uv run pytest                        # Run tests
uv run pytest path/to/test.py -v     # Single test file
uvx ruff check .                     # Lint
uvx ruff format .                    # Format
docker compose -f docker-compose.n8n.yml up -d  # Start n8n

Environment Variables

File Naming Conventions

n8n Workflow Patterns for Nathan

Detailed patterns and templates for creating n8n workflows in the Nathan project.

Complete Webhook Workflow Template

{
  "name": "Workflow Name",
  "nodes": [
    {
      "id": "webhook",
      "name": "Webhook",
      "type": "n8n-nodes-base.webhook",
      "typeVersion": 2,
      "position": [250, 300],
      "webhookId": "unique-webhook-id",
      "parameters": {
        "path": "webhook-path",
        "httpMethod": "POST",
        "responseMode": "responseNode"
      }
    },
    {
      "id": "validate-secret",
      "name": "Validate Secret",
      "type": "n8n-nodes-base.if",
      "typeVersion": 2,
      "position": [450, 300],
      "parameters": {
        "conditions": {
          "options": {
            "caseSensitive": true,
            "leftValue": "",
            "typeValidation": "strict"
          },
          "conditions": [
            {
              "id": "secret-check",
              "leftValue": "={{ $json.headers['x-n8n-secret'] }}",
              "rightValue": "={{ $env.N8N_WEBHOOK_SECRET }}",
              "operator": {
                "type": "string",
                "operation": "equals"
              }
            }
          ],
          "combinator": "and"
        }
      }
    },
    {
      "id": "unauthorized-response",
      "name": "Unauthorized Response",
      "type": "n8n-nodes-base.respondToWebhook",
      "typeVersion": 1.1,
      "position": [650, 450],
      "parameters": {
        "respondWith": "json",
        "responseBody": "={{ { success: false, error: \"Unauthorized\" } }}",
        "options": {
          "responseCode": 401
        }
      }
    },
    {
      "id": "operation",
      "name": "Main Operation",
      "type": "n8n-nodes-base.jira",
      "typeVersion": 1,
      "position": [650, 150],
      "parameters": {
        "resource": "issue",
        "operation": "get",
        "issueKey": "={{ $('Webhook').item.json.body.ticket_id }}"
      },
      "onError": "continueErrorOutput"
    },
    {
      "id": "success-response",
      "name": "Success Response",
      "type": "n8n-nodes-base.respondToWebhook",
      "typeVersion": 1.1,
      "position": [850, 100],
      "parameters": {
        "respondWith": "json",
        "responseBody": "={{ { success: true, data: $json } }}"
      }
    },
    {
      "id": "error-response",
      "name": "Error Response",
      "type": "n8n-nodes-base.respondToWebhook",
      "typeVersion": 1.1,
      "position": [850, 250],
      "parameters": {
        "respondWith": "json",
        "responseBody": "={{ { success: false, error: $json.error?.message || 'Operation failed' } }}",
        "options": {
          "responseCode": 500
        }
      }
    }
  ],
  "connections": {
    "Webhook": {
      "main": [[{"node": "Validate Secret", "type": "main", "index": 0}]]
    },
    "Validate Secret": {
      "main": [
        [{"node": "Main Operation", "type": "main", "index": 0}],
        [{"node": "Unauthorized Response", "type": "main", "index": 0}]
      ]
    },
    "Main Operation": {
      "main": [
        [{"node": "Success Response", "type": "main", "index": 0}],
        [{"node": "Error Response", "type": "main", "index": 0}]
      ]
    }
  },
  "settings": {
    "executionOrder": "v1",
    "meta": {
      "version": "1.0.0",
      "description": "Workflow description",
      "contract": {
        "input": {
          "param_name": "type (required|optional) - description"
        },
        "output": {
          "success": "boolean",
          "data": "object (on success)",
          "error": "string (on failure)"
        }
      }
    }
  }
}

Connection Patterns

Standard Flow (Success/Error Split)

"connections": {
  "Operation Node": {
    "main": [
      [{"node": "Success Handler", "type": "main", "index": 0}],
      [{"node": "Error Handler", "type": "main", "index": 0}]
    ]
  }
}

If Node Branching

Output 0 = TRUE branch, Output 1 = FALSE branch:

"connections": {
  "If Node": {
    "main": [
      [{"node": "True Branch", "type": "main", "index": 0}],
      [{"node": "False Branch", "type": "main", "index": 0}]
    ]
  }
}

AI Language Model Connection

"connections": {
  "LLM Model Node": {
    "ai_languageModel": [
      [{"node": "Chain/Agent Node", "type": "ai_languageModel", "index": 0}]
    ]
  }
}

Expression Patterns

Accessing Webhook Data

// Request body
$('Webhook').item.json.body.field_name

// Headers
$json.headers['x-custom-header']

// Query params (for GET requests)
$('Webhook').item.json.query.param_name

Referencing Other Nodes

// By node name
$('Node Name').item.json.field

// Current node input
$json.field

// Previous node in chain
$input.item.json.field

Array Operations (Properly Escaped)

// Map with string concatenation (not template literals)
$json.items.map(item => '"' + item + '"').join(',')

// Filter and map
$json.users.filter(u => u.active).map(u => u.name).join('\\n')

// Replace newlines
$json.text.replaceAll('\\n', ' ')

Code Node Pattern

For sandboxed code execution:

{
  "id": "execute-code",
  "name": "Execute Code",
  "type": "n8n-nodes-base.code",
  "typeVersion": 2,
  "position": [650, 200],
  "parameters": {
    "jsCode": "// Code here\nconst input = $('Webhook').item.json.body;\ntry {\n  const result = /* processing */;\n  return [{ json: { success: true, result } }];\n} catch (error) {\n  return [{ json: { success: false, error: error.message } }];\n}",
    "mode": "runOnceForAllItems"
  },
  "onError": "continueErrorOutput"
}

Position Guidelines

Standard grid positioning:

Spacing: 200px horizontal, 150px vertical between branches.

Credential References

{
  "credentials": {
    "jiraSoftwareCloudApi": {
      "id": "credential-id",
      "name": "Jira SW Cloud account"
    }
  }
}

Common credential types:

• jiraSoftwareCloudApi - Jira Cloud
• googleGeminiSdkApi - Google Gemini
• openAiApi - OpenAI
• slackApi - Slack

Metadata Contract

Always include a meta object in settings:

"settings": {
  "executionOrder": "v1",
  "meta": {
    "version": "1.0.0",
    "description": "What this workflow does",
    "contract": {
      "input": {
        "ticket_id": "string (required) - Jira issue key e.g. AOP-307"
      },
      "output": {
        "success": "boolean",
        "data": "object (on success) - Response data",
        "error": "string (on failure) - Error message"
      }
    }
  }
}

Python Patterns for Nathan

Detailed Python patterns and conventions for the Nathan project.

Module Template

"""Module description.

Detailed explanation of what this module does and its responsibilities.
"""

from __future__ import annotations

import logging
from dataclasses import dataclass
from pathlib import Path
from typing import Any

logger = logging.getLogger(__name__)


@dataclass(frozen=True, slots=True)
class DataModel:
    """Immutable data model with slots for memory efficiency."""

    name: str
    value: int
    optional_field: str | None = None


class ServiceClass:
    """Service class with clear responsibilities.

    Example:
        >>> service = ServiceClass()
        >>> result = service.do_something()
    """

    def __init__(self, config_path: Path | None = None) -> None:
        self.config_path = config_path or self._default_path()
        self._cache: dict[str, Any] = {}

    def _default_path(self) -> Path:
        return Path.home() / ".config" / "service.yaml"

    def do_something(self, param: str) -> dict[str, Any]:
        """Do something useful.

        Args:
            param: Description of parameter.

        Returns:
            Dictionary with result data.

        Raises:
            ValueError: If param is invalid.
        """
        if not param:
            raise ValueError("param cannot be empty")
        return {"result": param}


__all__ = ["DataModel", "ServiceClass"]

Async HTTP Client Pattern

"""Async HTTP client for webhook calls."""

from __future__ import annotations

from typing import Any

import httpx


async def trigger_webhook(
    *,
    url: str,
    parameters: dict[str, Any],
    timeout_s: float = 30.0,
    shared_secret: str | None = None,
    shared_secret_header: str = "X-N8N-SECRET",
) -> dict[str, Any]:
    """Execute webhook and return structured result.

    Returns a stable shape:
    - success: bool
    - data: dict (response JSON if any)
    - status_code: int | None
    - error: str | None
    """
    headers: dict[str, str] = {"Content-Type": "application/json"}
    if shared_secret:
        headers[shared_secret_header] = shared_secret

    try:
        async with httpx.AsyncClient(timeout=timeout_s) as client:
            resp = await client.post(url, json=parameters, headers=headers)
            status_code = resp.status_code
            resp.raise_for_status()

            try:
                data = resp.json()
            except ValueError:
                data = {"raw": resp.text}

            return {
                "success": True,
                "data": data,
                "status_code": status_code,
                "error": None,
            }

    except httpx.TimeoutException:
        return {
            "success": False,
            "data": {},
            "status_code": None,
            "error": f"Timeout after {timeout_s}s",
        }
    except httpx.HTTPStatusError as e:
        return {
            "success": False,
            "data": {},
            "status_code": e.response.status_code if e.response else None,
            "error": f"HTTP error: {e}",
        }
    except httpx.HTTPError as e:
        return {
            "success": False,
            "data": {},
            "status_code": None,
            "error": f"HTTP error: {e}",
        }

YAML Registry Pattern

"""YAML registry loader."""

from __future__ import annotations

from dataclasses import dataclass
from pathlib import Path
from typing import Any

import yaml


class RegistryError(RuntimeError):
    """Registry loading or validation error."""

    pass


@dataclass(frozen=True, slots=True)
class CommandDefinition:
    """Definition of a webhook command."""

    name: str
    endpoint: str
    method: str
    required_params: tuple[str, ...]
    optional_params: tuple[str, ...]
    description: str
    example: dict[str, Any]

    def validate_params(self, params: dict[str, Any]) -> None:
        """Validate required parameters are present."""
        missing = [k for k in self.required_params if k not in params]
        if missing:
            raise RegistryError(f"Missing required params: {missing}")


class Registry:
    """Load and manage command registry from YAML."""

    def __init__(self, config_path: Path) -> None:
        self.config_path = config_path
        self.commands: dict[str, CommandDefinition] = {}
        self._load()

    def _load(self) -> None:
        if not self.config_path.exists():
            return

        try:
            config = yaml.safe_load(self.config_path.read_text())
        except Exception as e:
            raise RegistryError(f"Failed to load YAML: {e}") from e

        if not config:
            return

        commands = config.get("commands", {})
        for name, cmd in commands.items():
            if not isinstance(cmd, dict):
                continue

            self.commands[name] = CommandDefinition(
                name=name,
                endpoint=str(cmd.get("endpoint", "")),
                method=str(cmd.get("method", "POST")),
                required_params=tuple(cmd.get("required_params", [])),
                optional_params=tuple(cmd.get("optional_params", [])),
                description=str(cmd.get("description", "")),
                example=dict(cmd.get("example", {})),
            )

    def get(self, name: str) -> CommandDefinition | None:
        return self.commands.get(name)

    def list_all(self) -> list[dict[str, Any]]:
        return [
            {
                "name": cmd.name,
                "description": cmd.description,
                "required_params": list(cmd.required_params),
                "example": cmd.example,
            }
            for cmd in sorted(self.commands.values(), key=lambda c: c.name)
        ]

Pydantic Model Pattern

"""Pydantic models for validation."""

from __future__ import annotations

from pydantic import BaseModel, Field


class NodeDefinition(BaseModel):
    """n8n node definition."""

    id: str = Field(..., description="Unique node identifier")
    name: str = Field(..., description="Display name")
    type: str = Field(..., description="Node type e.g. n8n-nodes-base.webhook")
    typeVersion: int = Field(default=1, ge=1)
    position: tuple[int, int] = Field(default=(0, 0))
    parameters: dict[str, object] = Field(default_factory=dict)
    webhookId: str | None = None
    credentials: dict[str, object] | None = None


class ConnectionDefinition(BaseModel):
    """Connection between nodes."""

    from_node: str = Field(..., alias="from")
    to_node: str = Field(..., alias="to")
    from_output: str = Field(default="main")
    from_index: int = Field(default=0, ge=0)
    to_input: str = Field(default="main")
    to_index: int = Field(default=0, ge=0)

    class Config:
        populate_by_name = True

Test Pattern

"""Test module for component."""

from __future__ import annotations

from pathlib import Path

import pytest

from nathan.module import Component


@pytest.fixture
def sample_data() -> dict[str, object]:
    """Provide sample test data."""
    return {"key": "value", "count": 42}


@pytest.fixture
def temp_config(tmp_path: Path) -> Path:
    """Create temporary config file."""
    config = tmp_path / "config.yaml"
    config.write_text("key: value\n")
    return config


class TestComponent:
    """Tests for Component class."""

    def test_init_default(self) -> None:
        """Test default initialization."""
        comp = Component()
        assert comp is not None

    def test_init_with_config(self, temp_config: Path) -> None:
        """Test initialization with config."""
        comp = Component(config_path=temp_config)
        assert comp.config_path == temp_config

    def test_process_valid_input(self, sample_data: dict[str, object]) -> None:
        """Test processing with valid input."""
        comp = Component()
        result = comp.process(sample_data)
        assert result["success"] is True

    def test_process_invalid_input(self) -> None:
        """Test processing with invalid input raises error."""
        comp = Component()
        with pytest.raises(ValueError, match="cannot be empty"):
            comp.process({})


@pytest.mark.asyncio
async def test_async_operation() -> None:
    """Test async operation."""
    result = await some_async_function()
    assert result is not None

Logging Pattern

"""Structured logging setup."""

from __future__ import annotations

import logging
import sys

def setup_logging(level: int = logging.INFO) -> None:
    """Configure structured logging."""
    logging.basicConfig(
        level=level,
        format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
        handlers=[logging.StreamHandler(sys.stderr)],
    )

# In modules, use module-level logger
logger = logging.getLogger(__name__)

# Log with context
logger.info("Processing item", extra={"item_id": item.id, "status": "started"})
logger.error("Operation failed", extra={"error": str(e)}, exc_info=True)

Type Hints Quick Reference