Agent Engineer

Agent Engineer

Role

You are an agent engineer. You extend the ob-engineer role with agent-specific implementation patterns using ol_ai_services.agent_dev_kit.

Read `skills/ob-engineer/SKILL.md` first and follow all of it. Then read skills/python-data-engineer/SKILL.md and skills/data-engineer/SKILL.md (ob-engineer's parents). This file contains only the additions and overrides that apply to agent building work.

---

Session Start — Verify Platform

Before implementing, confirm the agent platform:

1. Verify ol_ai_services is importable in the target project 2. Read skills/agent-architect/references/ol-ai-services-map.md for the correct import paths and available components 3. Read references/ob-library-selection.md (inherited from ob-engineer) to confirm the active OB variant 4. If ol_ai_services is not available, raise to the architect — do not implement without the platform

---

Additional References

---

Construction Order

Build agent components in this order (leaf-before-whole):

1. Common knowledge    — enums, types, constants for the agent domain
2. Tool implementations — BaseTool subclasses for each MISSING tool
3. Interop configs      — InteropServiceConfigs for each INTEROP tool
4. Agent configuration  — AgentConfiguration with model, tools, memory, constraints
5. Orchestration graph  — (if multi-agent) nodes, edges, conditional routes
6. Skill manifest       — (if packaging as skill) YAML + prompts + references
7. Runner / entry point — wire everything, create facade calls
8. Tests                — tool unit tests -> config tests -> graph tests -> E2E tests

---

Code Layout Convention

[agent_name]/
+-- common_knowledge/               # agent-domain enums, types, constants
|   +-- tool_name_enums.py          # registered tool name constants
|   +-- model_config_enums.py       # model name and parameter constants
|   +-- prompt_constants.py         # system prompt fragments as constants
+-- tools/                          # custom tool implementations
|   +-- [verb]_[subject]_tools.py   # one BaseTool subclass per file
|   +-- [verb]_[subject]_tool_inputs.py
|   +-- [verb]_[subject]_tool_outputs.py
+-- interop/                        # interop service configurations
|   +-- [service]_interop_configs.py
+-- configurations/                 # agent configurations
|   +-- [agent_name]_configurations.py
+-- orchestration/                  # multi-agent orchestration (if needed)
|   +-- [agent_name]_orchestration_graphs.py
+-- skills/                         # skill manifests (if packaging as skill)
|   +-- skill.yaml                  # SkillDefinition manifest
|   +-- prompts/
|   |   +-- system_prompt.md
|   |   +-- task_template.md
|   +-- references/
+-- runners/                        # entry points
|   +-- [agent_name]_runners.py
+-- tests/
    +-- test_tools/                 # tool unit tests
    +-- test_configurations/        # configuration validation tests
    +-- test_orchestration/         # graph validation tests
    +-- test_integration/           # agent E2E tests

---

Implementation Patterns

Step 1: Common Knowledge

All domain vocabulary as enums — no hardcoded strings in processing logic.

# tool_name_enums.py
from enum import Enum


class ToolNameEnums(
    Enum,
):
    SEARCH_DOCUMENTS = (
        "search_documents"
    )
    CREATE_ISSUE = (
        "create_issue"
    )

# model_config_enums.py
from enum import Enum


class ModelConfigEnums(
    Enum,
):
    DEFAULT_MODEL = (
        "claude-sonnet-4-6"
    )
    REASONING_MODEL = (
        "claude-opus-4-6"
    )

Step 2: Tool Implementation

Follow references/tool-implementation.md for the full BaseTool pattern.

Key rules:

• One tool class per file (OB convention: one public function per file)
• Class name: VerbSubjectTools (plural CamelCase, BORO naming)
• All parameters use named kwargs with * enforcement
• Input/output as Pydantic models with full type annotations
• Errors returned in output schema, never raised from _run()
• Register via ToolService.register_tool()

Step 3: Interop Configuration

Follow references/interop-implementation.md for service wiring.

Key rules:

• One config factory per external service
• Transport selection: MCP for standardized tools, REST for legacy APIs, DIRECT for Python libs
• Configuration values from enums, not hardcoded strings
• Auth config from environment variables (read once at entry point)

Step 4: Agent Configuration

Follow references/agent-implementation.md for configuration patterns.

Key rules:

• Factory function returns AgentConfiguration (not direct instantiation in runner)
• Model name from ModelConfigEnums
• System prompt as constant or loaded from file
• Memory config matches architect's design
• All named parameters with * enforcement

Step 5: Orchestration Graph (if multi-agent)

Follow references/agent-implementation.md for graph patterns.

Key rules:

• Graph must be a valid DAG (no cycles)
• Each node maps to an existing AgentConfiguration
• Conditional routes use on_status for edge evaluation
• Entry node explicitly declared

Step 6: Skill Manifest (if packaging as skill)

Follow references/skill-manifest.md for YAML manifest creation.

Key rules:

• Manifest metadata triggers auto-discovery (description must be specific)
• Progressive disclosure: L1 metadata, L2 prompts, L3 references
• Input/output schemas as JSON Schema
• Evaluation queries for testing (10+ cases)

Step 7: Runner / Entry Point

Follow references/agent-implementation.md for runner wiring.

Key rules:

• Environment variables read once at this level only
• Custom tools registered before agent creation
• Interop tools discovered and registered before agent creation
• Facade used for all lifecycle operations
• Thread ID passed through for conversation continuity

Step 8: Tests

Follow construction order for tests:

---

Sub-Skill Delegation

---

Verification Checklist

After implementation, verify:

• [ ] All custom tools extend BaseTool with proper Pydantic schemas
• [ ] All tools registered via ToolService (not ad-hoc instantiation)
• [ ] Interop service configs only in interop/ directory
• [ ] Agent configuration uses enum constants, not hardcoded strings
• [ ] System prompt stored as constant or file, not inline in runner
• [ ] Memory configuration matches architect's design
• [ ] Orchestration graph is a valid DAG (no cycles)
• [ ] Each tool independently testable
• [ ] No module-level mutable state
• [ ] All parameters use named kwargs with * enforcement (OB convention)
• [ ] All type annotations present on params and returns (OB convention)
• [ ] Class names plural CamelCase (OB convention)
• [ ] One public function per file (OB convention)
• [ ] Private methods use __double_underscore (OB convention)
• [ ] No hardcoded strings — all in enums/constants (OB convention)
• [ ] Environment variables read once at runner level only

---

Quality Gates

ruff check src/                    # linting
ruff format src/                   # formatting (20-char line discipline via review)
mypy src/ --strict                 # type checking in strict mode
pytest tests/test_tools/           # tool unit tests pass
pytest tests/test_configurations/  # config validation tests pass
pytest tests/test_orchestration/   # graph validation tests pass
pytest tests/test_integration/     # E2E tests pass

---

Review Mode

When reviewing existing agent code, check against:

Severity classification:

• CRITICAL: Tools not registered (unreusable); no BaseTool contract (unresolvable); circular dependencies
• MAJOR: Missing tests; ad-hoc interop; no constants layer; missing type annotations
• MINOR: Naming inconsistencies; suboptimal construction order; loose memory config

---

Feedback

If the user corrects this skill's output due to a misinterpretation or missing rule in the skill itself (not a one-off preference), invoke skill-feedback to capture structured feedback and optionally post a GitHub issue.

If skill-feedback is not installed, ask the user: "This looks like a skill defect. Would you like to install the `skill-feedback` skill to report it?" If the user declines, continue without feedback capture.

Agent Implementation Guide

Detailed patterns for agent configuration, orchestration graph building, and runner wiring using ol_ai_services.

---

1. Agent Configuration Patterns

Simple Agent (Single Model, Tools Only)

from ol_ai_services.agent_dev_kit.objects.agent_configuration import (
    AgentConfiguration,
)
from ol_ai_services.agent_dev_kit.objects.agent_configuration import (
    StorageStrategy,
)
from ..common_knowledge.model_config_enums import (
    ModelConfigEnums,
)
from ..common_knowledge.prompt_constants import (
    SIMPLE_AGENT_SYSTEM_PROMPT,
)


def create_simple_agent_configuration(
    *,
    workspace_id: str,
    tool_ids: list[str],
) -> AgentConfiguration:
    return AgentConfiguration(
        workspace_id=workspace_id,
        name="simple_agent",
        description=(
            "A simple tool-using agent"
        ),
        model_name=(
            ModelConfigEnums
            .DEFAULT_MODEL
            .value
        ),
        system_prompt=(
            SIMPLE_AGENT_SYSTEM_PROMPT
        ),
        tool_ids=tool_ids,
        backend_storage=(
            StorageStrategy.EPHEMERAL
        ),
    )

Memory-Augmented Agent

from ol_ai_services.agent_dev_kit.objects.memory_configuration import (
    MemoryConfiguration,
)


def create_memory_agent_configuration(
    *,
    workspace_id: str,
    tool_ids: list[str],
) -> AgentConfiguration:
    return AgentConfiguration(
        workspace_id=workspace_id,
        name="memory_agent",
        description=(
            "Agent with recall "
            "and persistence"
        ),
        model_name=(
            ModelConfigEnums
            .DEFAULT_MODEL
            .value
        ),
        system_prompt=(
            MEMORY_AGENT_SYSTEM_PROMPT
        ),
        tool_ids=tool_ids,
        backend_storage=(
            StorageStrategy.PERSISTENT
        ),
        memory_config=MemoryConfiguration(
            engine_type="vector",
            recall_enabled=True,
            max_recall_results=5,
            max_recall_tokens=500,
        ),
    )

Agent with Human-in-the-Loop

def create_supervised_agent_configuration(
    *,
    workspace_id: str,
    tool_ids: list[str],
) -> AgentConfiguration:
    return AgentConfiguration(
        workspace_id=workspace_id,
        name="supervised_agent",
        description=(
            "Agent requiring approval "
            "for destructive actions"
        ),
        model_name=(
            ModelConfigEnums
            .DEFAULT_MODEL
            .value
        ),
        system_prompt=(
            SUPERVISED_AGENT_SYSTEM_PROMPT
        ),
        tool_ids=tool_ids,
        interrupt_config={
            "require_approval": [
                "delete_record",
                "send_email",
                "modify_config",
            ],
        },
    )

---

2. Orchestration Graph Building

Linear Pipeline

from ol_ai_services.agent_dev_kit.objects.orchestration_graph import (
    OrchestrationGraph,
    AgentNode,
    AgentEdge,
)


def create_pipeline_graph(
    *,
    workspace_id: str,
    collect_agent_id: str,
    transform_agent_id: str,
    output_agent_id: str,
) -> OrchestrationGraph:
    return OrchestrationGraph(
        workspace_id=workspace_id,
        name="data_pipeline",
        description=(
            "Linear data processing "
            "pipeline"
        ),
        nodes=[
            AgentNode(
                node_id="collect",
                agent_id=collect_agent_id,
                input_mapping={
                    "source": "input.source",
                },
            ),
            AgentNode(
                node_id="transform",
                agent_id=transform_agent_id,
                input_mapping={
                    "data": "collect.output",
                },
            ),
            AgentNode(
                node_id="output",
                agent_id=output_agent_id,
                input_mapping={
                    "data": (
                        "transform.output"
                    ),
                },
            ),
        ],
        edges=[
            AgentEdge(
                from_node_id="collect",
                to_node_id="transform",
            ),
            AgentEdge(
                from_node_id="transform",
                to_node_id="output",
            ),
        ],
        entry_node_id="collect",
    )

Conditional Router

def create_router_graph(
    *,
    workspace_id: str,
    router_agent_id: str,
    research_agent_id: str,
    coding_agent_id: str,
    data_agent_id: str,
) -> OrchestrationGraph:
    return OrchestrationGraph(
        workspace_id=workspace_id,
        name="task_router",
        description=(
            "Routes tasks to "
            "specialist agents"
        ),
        nodes=[
            AgentNode(
                node_id="router",
                agent_id=(
                    router_agent_id
                ),
            ),
            AgentNode(
                node_id="research",
                agent_id=(
                    research_agent_id
                ),
            ),
            AgentNode(
                node_id="coding",
                agent_id=(
                    coding_agent_id
                ),
            ),
            AgentNode(
                node_id="data",
                agent_id=(
                    data_agent_id
                ),
            ),
        ],
        edges=[
            AgentEdge(
                from_node_id="router",
                to_node_id="research",
                conditional_route={
                    "on_status": (
                        "research"
                    ),
                },
            ),
            AgentEdge(
                from_node_id="router",
                to_node_id="coding",
                conditional_route={
                    "on_status": (
                        "coding"
                    ),
                },
            ),
            AgentEdge(
                from_node_id="router",
                to_node_id="data",
                conditional_route={
                    "on_status": "data",
                },
            ),
        ],
        entry_node_id="router",
    )

Peer Review (Parallel + Merge)

def create_peer_review_graph(
    *,
    workspace_id: str,
    reviewer_a_id: str,
    reviewer_b_id: str,
    synthesizer_id: str,
) -> OrchestrationGraph:
    return OrchestrationGraph(
        workspace_id=workspace_id,
        name="peer_review",
        description=(
            "Parallel review with "
            "synthesis"
        ),
        nodes=[
            AgentNode(
                node_id="reviewer_a",
                agent_id=reviewer_a_id,
            ),
            AgentNode(
                node_id="reviewer_b",
                agent_id=reviewer_b_id,
            ),
            AgentNode(
                node_id="synthesizer",
                agent_id=synthesizer_id,
                input_mapping={
                    "review_a": (
                        "reviewer_a.output"
                    ),
                    "review_b": (
                        "reviewer_b.output"
                    ),
                },
            ),
        ],
        edges=[
            AgentEdge(
                from_node_id="reviewer_a",
                to_node_id="synthesizer",
            ),
            AgentEdge(
                from_node_id="reviewer_b",
                to_node_id="synthesizer",
            ),
        ],
        entry_node_id="reviewer_a",
    )

---

3. Runner Wiring

Full Runner Template

# [agent_name]_runners.py
import os

from ol_ai_services.agent_dev_kit.facade import (
    AgentDevelopmentKitFacade,
)
from ..common_knowledge.tool_name_enums import (
    ToolNameEnums,
)
from ..tools.verb_subject_tools import (
    VerbSubjectTools,
)
from ..interop.service_interop_configs import (
    create_service_interop_config,
)
from ..configurations.agent_configurations import (
    create_agent_configuration,
)


async def run_agent(
    *,
    facade: AgentDevelopmentKitFacade,
    workspace_id: str,
    input_message: str,
    thread_id: str | None = None,
) -> dict:
    # 1. Read environment config
    #    (once, at entry point)
    service_endpoint = os.environ[
        "SERVICE_MCP_ENDPOINT"
    ]

    # 2. Register custom tools
    custom_tool = VerbSubjectTools()
    await facade.register_tool(
        tool=custom_tool,
    )

    # 3. Connect interop and
    #    discover tools
    interop_config = (
        create_service_interop_config(
            endpoint=service_endpoint,
        )
    )
    interop_tools = (
        await facade
        .discover_interop_tools(
            config=interop_config,
            workspace_id=workspace_id,
        )
    )

    # 4. Collect all tool IDs
    tool_ids = [
        ToolNameEnums
        .VERB_SUBJECT
        .value,
    ] + [
        tool.name
        for tool in interop_tools
    ]

    # 5. Create agent configuration
    agent_config = (
        create_agent_configuration(
            workspace_id=workspace_id,
            tool_ids=tool_ids,
        )
    )
    agent_id = await facade.create_agent(
        configuration=agent_config,
    )

    # 6. Execute
    execution = (
        await facade.execute_agent(
            agent_id=agent_id,
            input_data={
                "message": input_message,
            },
            thread_id=thread_id,
        )
    )

    return execution

Runner Rules

1. Environment variables read here only — never in tools, configs, or orchestration 2. Tool registration before agent creation — all tools must be registered first 3. Interop discovery before agent creation — discover and register external tools 4. Facade for all lifecycle ops — do not call internal services directly 5. Thread ID for continuity — pass through for multi-turn conversations 6. No business logic in runner — runner is wiring only, like a bclearer pipeline runner

---

4. Testing Patterns

Tool Unit Test

import pytest

from ..tools.verb_subject_tools import (
    VerbSubjectTools,
)


@pytest.fixture
def tool() -> VerbSubjectTools:
    return VerbSubjectTools()


@pytest.mark.asyncio
async def test_happy_path(
    tool: VerbSubjectTools,
) -> None:
    result = await tool.run(
        param_one="test_value",
    )
    assert result.success is True
    assert result.result != ""


@pytest.mark.asyncio
async def test_error_handling(
    tool: VerbSubjectTools,
) -> None:
    result = await tool.run(
        param_one="",
    )
    assert result.success is False
    assert result.error_message != ""

Configuration Validation Test

from ..configurations.agent_configurations import (
    create_agent_configuration,
)


def test_config_is_valid() -> None:
    config = create_agent_configuration(
        workspace_id="test-ws",
        tool_ids=[
            "tool_a",
            "tool_b",
        ],
    )
    assert config.name != ""
    assert config.model_name != ""
    assert len(config.tool_ids) == 2

Orchestration Graph Test

from ..orchestration.agent_orchestration_graphs import (
    create_pipeline_graph,
)


def test_graph_is_valid_dag() -> None:
    graph = create_pipeline_graph(
        workspace_id="test-ws",
        collect_agent_id="a1",
        transform_agent_id="a2",
        output_agent_id="a3",
    )
    assert (
        graph.entry_node_id
        == "collect"
    )
    assert len(graph.nodes) == 3
    assert len(graph.edges) == 2
    # Verify no cycles:
    # all to_node_ids come after
    # from_node_ids in topological order
    node_order = {
        node.node_id: i
        for i, node
        in enumerate(graph.nodes)
    }
    for edge in graph.edges:
        assert (
            node_order[
                edge.from_node_id
            ]
            < node_order[
                edge.to_node_id
            ]
        )

Integration Test

@pytest.mark.asyncio
async def test_agent_end_to_end() -> None:
    facade = (
        AgentDevelopmentKitFacade(...)
    )
    result = await run_agent(
        facade=facade,
        workspace_id="test-ws",
        input_message="test query",
    )
    assert (
        result["status"]
        == "COMPLETED"
    )

Interop Implementation Guide

How to configure and use interop clients for MCP and REST services in ol_ai_services.

---

1. InteropServiceConfigs

Every external service connection is defined by an InteropServiceConfigs object:

from ol_ai_services.agent_dev_kit.objects.interop_service_configs import (
    InteropServiceConfigs,
)
from ol_ai_services.agent_dev_kit.objects.transport_types import (
    TransportTypes,
)
from ol_ai_services.agent_dev_kit.objects.interop_service_configs import (
    RetryConfigs,
)


def create_service_interop_config(
    *,
    endpoint: str,
    auth_token: str | None = None,
) -> InteropServiceConfigs:
    return InteropServiceConfigs(
        service_name=(
            "Human-Readable "
            "Service Name"
        ),
        transport=TransportTypes.MCP,
        endpoint=endpoint,
        auth_config=(
            {"bearer_token": auth_token}
            if auth_token
            else None
        ),
        tool_prefix="service",
        timeout_seconds=30,
        discovery_enabled=True,
        retry_config=RetryConfigs(
            max_retries=3,
            backoff_factor=2,
        ),
    )

---

2. MCP Client Usage

Stdio Transport (Local MCP Server)

config = InteropServiceConfigs(
    service_name=(
        "Local Document Search"
    ),
    transport=TransportTypes.MCP,
    endpoint=(
        "npx -y "
        "@myorg/document-search-mcp"
    ),
    discovery_enabled=True,
)

SSE Transport (Remote MCP Server)

config = InteropServiceConfigs(
    service_name=(
        "Remote Document Search"
    ),
    transport=TransportTypes.MCP,
    endpoint=(
        "https://mcp.example.com/sse"
    ),
    auth_config={
        "bearer_token": token,
    },
    discovery_enabled=True,
)

Tool Discovery and Registration

from ol_ai_services.agent_dev_kit.interop.interop_client_factories import (
    InteropClientFactories,
)

factory = InteropClientFactories()

async with (
    factory.create_and_connect(
        config=config,
    ) as client
):
    # Discover all tools from
    # the service
    discovered_tools = (
        await client.discover_tools()
    )

    # Register each discovered tool
    for tool in discovered_tools:
        await tool_service.register_tool(
            tool=tool,
            workspace_id=workspace_id,
        )

    # Or invoke a specific tool
    # directly
    result = (
        await client.invoke_tool(
            tool_name=(
                "search_documents"
            ),
            query=(
                "agent architecture"
            ),
            max_results=10,
        )
    )

---

3. REST Client Usage

config = InteropServiceConfigs(
    service_name="Legacy API",
    transport=TransportTypes.REST,
    endpoint=(
        "https://api.example.com"
    ),
    auth_config={
        "bearer_token": token,
    },
    timeout_seconds=60,
    retry_config=RetryConfigs(
        max_retries=3,
        backoff_factor=2,
    ),
)

REST clients support:

• Automatic retry with exponential backoff
• Bearer token authentication
• Tool discovery via REST endpoints (if service supports it)
• Direct tool invocation via HTTP

---

4. Direct (Python Import) Client

config = InteropServiceConfigs(
    service_name=(
        "Internal Library"
    ),
    transport=TransportTypes.DIRECT,
    endpoint=(
        "mypackage.tools"
        ".MyToolClass"
    ),
    discovery_enabled=False,
)

Use DIRECT transport for tools that are Python packages in the same environment. No network calls, no serialization overhead.

---

5. Configuration from Environment

Following OB conventions, environment variables are read once at the entry point:

# In runner (entry point only):
import os


service_endpoint = os.environ[
    "DOCUMENT_SEARCH_MCP_ENDPOINT"
]
auth_token = os.environ.get(
    "DOCUMENT_SEARCH_AUTH_TOKEN",
)

config = (
    create_document_search_interop_config(
        endpoint=service_endpoint,
        auth_token=auth_token,
    )
)

Rules:

• Environment variables read once at the entry point (runner)
• Values stored in configuration objects, not passed as function args between modules
• No os.environ access in tools, configurations, or orchestration modules
• All endpoints must be absolute URLs or valid stdio commands by the time they enter config

---

6. MCP Tool Factory (Dynamic Tool Generation)

When MCP tools are discovered, MCPToolFactories generates BaseTool subclasses at runtime:

from ol_ai_services.agent_dev_kit.interop.mcp_tool_factories import (
    MCPToolFactories,
)

# Automatic — happens inside
# MCPInteropClients.discover_tools()
#
# You rarely call this directly;
# use discover_tools() instead.

What it does: 1. Reads MCP tool definition (name, description, JSON Schema) 2. Converts JSON Schema to Pydantic input_schema and output_schema 3. Generates async _run() method bound to the client callback 4. Creates a BaseTool subclass at runtime via type() and create_model()

The generated tools conform to the same BaseTool contract as hand-written tools and can be registered with ToolService identically.

---

7. Naming Conventions

---

8. Implementing a New MCP Server

When the architect's design calls for a new MCP server (not just a client):

Python (using FastMCP or mcp package)

# [service]_mcp/server.py
from mcp.server import Server
from mcp.types import Tool

server = Server(
    name="service-mcp",
)


@server.tool()
async def search_documents(
    *,
    query: str,
    max_results: int = 10,
) -> str:
    """Search documents by
    semantic query.

    Use when the user asks about
    existing documentation.
    Returns max 10 results.
    """
    # Implementation
    results = (
        await __execute_search(
            query=query,
            max_results=max_results,
        )
    )
    return results

TypeScript (using @modelcontextprotocol/sdk)

import { McpServer } from
  "@modelcontextprotocol/sdk/server/mcp.js";

const server = new McpServer({
  name: "service-mcp-server",
  version: "1.0.0",
});

server.tool(
  "search_documents",
  {
    query: z.string(),
    max_results: z.number()
      .default(10),
  },
  async ({ query, max_results }) => {
    // Implementation
  },
);

MCP Server Rules

1. Tool naming: snake_case, action-oriented verbs 2. Tool descriptions: Same quality rules as BaseTool descriptions 3. Pagination: Implement with has_more, next_offset/next_cursor, total_count 4. Annotations: Provide readOnlyHint, destructiveHint, idempotentHint 5. Error format: Structured JSON with code, message, details 6. Transport: stdio for local, SSE for remote

Skill Manifest Guide

How to create SkillDefinition YAML manifests for packaging agents as reusable skills in ol_ai_services.

---

1. Manifest Structure

apiVersion: ol.ai/v1
kind: Skill
metadata:
    name: [skill-name]
    version: "1.0"
    category: [prompt_category]
    description: >
        Brief description of what this skill does and when to use it.
        Include trigger conditions for auto-invocation.
spec:
    model:
        default: "claude-sonnet-4-6"
        allowed:
            - "claude-sonnet-4-6"
            - "claude-opus-4-6"
    input_schema:
        type: object
        properties:
            query:
                type: string
                description: "The user's request"
        required:
            - query
    output_schema:
        type: object
        properties:
            result:
                type: string
                description: "The skill's output"
    tools:
        - name: [tool_name]
          description: "What this tool does"
          source: PACKAGE
          implementation_class: "module.path.ToolClass"
        # For INTEROP tools:
        # - name: [tool_name]
        #   source: INTEROP
        #   service: "service_name"
    prompts:
        system: prompts/system_prompt.md
        task: prompts/task_template.md
        references:
            - references/domain_knowledge.md
            - references/examples.md
    output_template: |
        ## Result
        {{ result }}
    execution:
        timeout_seconds: 120
        max_llm_calls: 10
        max_tool_calls: 20
        memory:
            engine_type: "vector"
            recall_enabled: true
            max_recall_results: 5
            max_recall_tokens: 500

---

2. File Layout

skills/
+-- [skill-name]/
    +-- skill.yaml              # SkillDefinition manifest
    +-- prompts/
    |   +-- system_prompt.md    # Agent identity + constraints
    |   +-- task_template.md    # Task prompt with {{ variables }}
    +-- references/             # Domain knowledge loaded as context
        +-- [topic].md
        +-- ...

---

3. Prompt Files

System Prompt

# [Skill Name]

You are a [role description].

## Capabilities
- [capability 1]
- [capability 2]

## Constraints
- [constraint 1]
- [constraint 2]

## Output Format
[Expected output structure]

Task Template (Jinja2)

## Task

{{ query }}

{% if context %}
## Context

{{ context }}
{% endif %}

## Instructions

1. [step 1]
2. [step 2]
3. [step 3]

---

4. Progressive Disclosure

Apply three-level progressive disclosure:

Description Optimization

The manifest description is the primary trigger for auto-invocation by the SkillRegistry. It must be specific enough to trigger correctly:

# BAD — too vague, will undertrigger
description: "Helps with documents"

# GOOD — specific triggers and scope
description: >
    Searches, summarizes, and answers questions about project documentation.
    Use when: user asks about existing docs, needs a doc summary, wants to
    find information across multiple documents. Supports PDF, Markdown, and
    HTML sources.

Description Checklist

• [ ] States what the skill does (actions)
• [ ] States when to use it (trigger conditions)
• [ ] States what it supports (scope/formats)
• [ ] Does NOT trigger on unrelated requests

---

5. Skill Registration

Skills are discovered by SkillRegistry from the skills directory:

# Automatic discovery:
# Place skill.yaml in the skills directory.
# SkillRegistry.discover_skills() finds it.

# Manual invocation:
result = (
    await skill_execution_service
    .invoke_skill(
        skill_name="skill-name",
        input_data={
            "query": "user request",
        },
        workspace_id=workspace_id,
    )
)

Skill Execution Pipeline

1. Validate input against input_schema (jsonschema) 2. Select model (default or override, checked against allowed list) 3. Load prompts (system + task), render Jinja2 templates 4. Assemble messages (system message with references + user message) 5. Resolve tools (PACKAGE via import, INTEROP via client connect + discover) 6. Execute transient agent with timeout via AgentExecutionRuntime 7. Render output template, validate against output_schema 8. Return result with metrics (duration, LLM calls, tool calls)

---

6. Evaluation

Design 10+ evaluation queries to verify the skill triggers correctly and produces quality output:

evaluations:
    # Should-trigger cases (verify recall)
    - query: "[realistic user request that should invoke this skill]"
      expected: "[what the output should contain]"
      should_trigger: true

    - query: "[another valid request, different phrasing]"
      expected: "[expected output]"
      should_trigger: true

    # Should-NOT-trigger cases (verify precision)
    - query: "[request that looks similar but is out of scope]"
      should_trigger: false

    - query: "[completely unrelated request]"
      should_trigger: false

Evaluation Design Rules

• Mix of should-trigger (60%) and should-not-trigger (40%)
• Should-trigger cases use varied phrasing (not just rewording the description)
• Should-not-trigger cases include near-misses (similar domain, different task)
• Expected outputs are specific enough to grade programmatically

---

7. Manifest Naming Conventions

---

8. Manifest Validation Checklist

Before registering:

• [ ] metadata.name is unique across the skill registry
• [ ] metadata.description is specific with clear trigger conditions
• [ ] spec.input_schema covers all required parameters
• [ ] spec.output_schema matches what the skill actually produces
• [ ] spec.tools lists all tools the skill needs (source type correct)
• [ ] spec.prompts.system file exists and is well-structured
• [ ] spec.prompts.task file exists with correct Jinja2 variables
• [ ] spec.execution.timeout_seconds is realistic for the task
• [ ] spec.execution.max_llm_calls prevents runaway execution
• [ ] Evaluation queries written (10+ cases, mixed trigger/no-trigger)
• [ ] All referenced files exist at their declared paths

Tool Implementation Guide

How to implement tools for registration into ol_ai_services using the BaseTool contract.

---

1. BaseTool Contract

Every tool inherits from ol_ai_services.agent_dev_kit.tools.base_tool.BaseTool:

from abc import ABC, abstractmethod
from typing import Type
from pydantic import BaseModel


class BaseTool(ABC):
    name: str
    description: str
    input_schema: Type[BaseModel]
    output_schema: Type[BaseModel]

    @abstractmethod
    async def _run(
        self,
        *,
        input_data: BaseModel,
    ) -> BaseModel:
        ...

    async def run(
        self,
        **kwargs,
    ) -> BaseModel:
        # Validates input against
        # input_schema, calls _run(),
        # handles errors
        ...

---

2. Implementation Template

Step 1: Define Input Schema

# [verb]_[subject]_tool_inputs.py
from pydantic import (
    BaseModel,
    Field,
)


class VerbSubjectToolInputs(
    BaseModel,
):
    """Input schema for
    [tool description]."""

    param_one: str = Field(
        description=(
            "What this parameter "
            "controls"
        ),
    )
    param_two: int = Field(
        default=10,
        description=(
            "Optional parameter "
            "with default"
        ),
    )

Step 2: Define Output Schema

# [verb]_[subject]_tool_outputs.py
from pydantic import (
    BaseModel,
    Field,
)


class VerbSubjectToolOutputs(
    BaseModel,
):
    """Output schema for
    [tool description]."""

    result: str = Field(
        description=(
            "The primary result"
        ),
    )
    success: bool = Field(
        default=True,
        description=(
            "Whether the operation "
            "succeeded"
        ),
    )
    error_message: str = Field(
        default="",
        description=(
            "Error details if "
            "success is false"
        ),
    )

Step 3: Implement Tool

# [verb]_[subject]_tools.py
from ol_ai_services.agent_dev_kit.tools.base_tool import (
    BaseTool,
)
from .verb_subject_tool_inputs import (
    VerbSubjectToolInputs,
)
from .verb_subject_tool_outputs import (
    VerbSubjectToolOutputs,
)


class VerbSubjectTools(
    BaseTool,
):
    name: str = "verb_subject"
    description: str = (
        "Clear description of what "
        "this tool does and when "
        "to use it."
    )
    input_schema = (
        VerbSubjectToolInputs
    )
    output_schema = (
        VerbSubjectToolOutputs
    )

    async def _run(
        self,
        *,
        input_data: (
            VerbSubjectToolInputs
        ),
    ) -> VerbSubjectToolOutputs:
        try:
            result = (
                await self
                .__do_the_work(
                    param_one=(
                        input_data
                        .param_one
                    ),
                    param_two=(
                        input_data
                        .param_two
                    ),
                )
            )
        except SpecificError as error:
            return (
                VerbSubjectToolOutputs(
                    result="",
                    success=False,
                    error_message=(
                        str(error)
                    ),
                )
            )
        return VerbSubjectToolOutputs(
            result=result,
            success=True,
        )

    async def __do_the_work(
        self,
        *,
        param_one: str,
        param_two: int,
    ) -> str:
        # Private implementation
        ...

Step 4: Register Tool

from ol_ai_services.agent_dev_kit.tools.tool_service import (
    ToolService,
)


# In the runner (entry point):
tool = VerbSubjectTools()
await tool_service.register_tool(
    tool=tool,
    workspace_id=workspace_id,
)

---

3. Naming Conventions (OB/BORO)

---

4. Tool Design Principles

Description Quality

Tool descriptions are critical — they are the agent's only API documentation:

• Lead with what it does: "Searches documents in the knowledge base by semantic query"
• Include when to use: "Use when the user asks about existing documentation"
• Mention key parameters: "Requires a query string; optionally filters by date range"
• Note limitations: "Returns max 10 results; does not search attachments"

Schema Design

• Required vs optional: Mark truly required params; use defaults for optional
• Field descriptions: Every field has a description (the agent reads these)
• Constrained types: Use Field(ge=1, le=100) for bounded values
• Enum fields: Use Literal["option_a", "option_b"] for fixed choices
• No bare `dict` or `Any`: Use typed models or dict[str, str] at minimum

Error Handling

Tools return errors in the output schema — they do not raise exceptions from _run(). The agent needs structured error information to decide what to do next.

Every output schema must include:

• success: bool — whether the operation succeeded
• error_message: str — details if success is False

Idempotency

If a tool is idempotent, document it. If not, the agent needs to know to avoid duplicate calls. Include this in the tool description.

---

5. Testing Tools

Unit Test Template

# test_verb_subject_tools.py
import pytest

from [module].verb_subject_tools import (
    VerbSubjectTools,
)


@pytest.fixture
def tool() -> VerbSubjectTools:
    return VerbSubjectTools()


@pytest.mark.asyncio
async def test_run_with_valid_input(
    tool: VerbSubjectTools,
) -> None:
    result = await tool.run(
        param_one="test_value",
    )
    assert result.success is True
    assert result.result != ""


@pytest.mark.asyncio
async def test_run_returns_error(
    tool: VerbSubjectTools,
) -> None:
    # Test with input that triggers
    # error path
    result = await tool.run(
        param_one="invalid",
    )
    assert result.success is False
    assert result.error_message != ""


@pytest.mark.asyncio
async def test_schema_validation(
    tool: VerbSubjectTools,
) -> None:
    # Verify input schema rejects
    # invalid types
    with pytest.raises(
        ValidationError,
    ):
        await tool.run(
            param_one=123,  # wrong type
        )

Test Checklist

• [ ] Happy path returns expected output
• [ ] Error path returns structured error (not exception)
• [ ] Input schema rejects invalid types
• [ ] Required parameters are enforced
• [ ] Optional parameters use defaults
• [ ] Output schema is complete (all fields populated)
• [ ] Tool is stateless (multiple calls produce independent results)