skillchain 0.2.1

Orchestrate AI skills in composable patterns

SkillChain

Orchestrate AI skills in composable patterns.

SkillChain is a Python SDK for chaining AI skills together — each potentially using a different LLM model — through sequential, parallel, conditional, map-reduce, and loop patterns. It follows the agentskills.io specification and provides built-in OpenTelemetry tracing with GenAI semantic conventions.

SkillChain orchestrates skills — portable, spec-compliant units of procedural knowledge that any agentskills.io-compatible tool can load and execute.

---

Table of Contents

• Installation
• Quick Start
• Architecture
  • Core Abstractions
  • How Context Flows Between Skills
  • Progressive Disclosure
  • Package Structure
• Orchestration Patterns
  • Sequential
  • Parallel
  • Conditional
  • MapReduce
  • Loop
  • Composing Patterns
• Defining Skills
  • Decorator Style
  • Class Style
  • From SKILL.md Directory
• Built-in Skills
• Skill Registry
• Per-Skill Model Selection
• OpenTelemetry Tracing
• Examples
  • Example 1: Simple Chain with Claude
  • Example 2: Parallel Analysis
  • Example 3: Document Pipeline
  • Example 4: Conditional Routing
  • Example 5: All Patterns Combined
  • Example 6: Custom Skills from Directories
• Error Handling
• License

---

Installation

# Core SDK
pip install skillchain

# With OpenTelemetry tracing support
pip install skillchain[telemetry]

Dependencies: LiteLLM (100+ model support), PyYAML, httpx, Pydantic

Python: >= 3.10

---

Quick Start

from skillchain import skill

# Define skills — each can use a different model
@skill(name="summarize", model="claude-sonnet-4-6")
async def summarize(ctx):
    return f"Summarize this text in 2 sentences:\n\n{ctx['text']}"

@skill(name="translate", model="claude-haiku-4-5-20251001")
async def translate(ctx):
    return f"Translate to French:\n\n{ctx['summarize']}"

# Chain them with >>
chain = summarize >> translate

# Run (sync wrapper for scripts/notebooks)
result = chain.run_sync({"text": "Long article about AI..."})
print(result["summarize"])   # English summary
print(result["translate"])   # French translation

Set your API key first:

export ANTHROPIC_API_KEY="your-key-here"

---

Architecture

Core Abstractions

SkillChain is built on four core abstractions:

+------------------+     +------------------+     +------------------+
|      Skill       |     |   SkillContext    |     | ExecutionEngine  |
|------------------|     |------------------|     |------------------|
| name             |     | _data (dict)     |     | provider         |
| description      |     | history          |     |                  |
| model            |---->| results          |<----|  execute()       |
| instructions     |     |                  |     |    |             |
|                  |     | get/set/merge    |     |    v             |
| run(ctx) --------|---->| snapshot()       |     | ModelProvider    |
| build_prompt()   |     | record()         |     |   call() -----> LLM
| process_output() |     +------------------+     +------------------+
+------------------+
        |
        | implements
        v
+------------------+
|      Chain       |  (Chain is itself a Skill — composite pattern)
|------------------|
| _skills: [Skill] |
| run(ctx) loops   |
|   each skill     |
| >> operator      |
+------------------+

Skill — The atomic unit. Wraps a prompt template + model config. Two ways to define: @skill decorator or Skill subclass.

SkillContext — A dict-like object that flows through the chain. Every skill reads from it and writes to it. Tracks execution history and intermediate results.

ExecutionEngine — Runs a skill's prompt through LiteLLM (or returns it directly for local skills with model=None).

Chain — A composed sequence of skills. Built via the >> operator. A Chain is itself a Skill, so chains nest inside chains.

How Context Flows Between Skills

The SkillContext is the single shared data bus. Here is exactly what happens when a skill runs:

                         SkillContext
                    +--------------------+
                    | "text": "Hello..." |
                    | "language": "FR"   |
                    +--------------------+
                              |
                    +---------v----------+
                    |   summarize.run()  |
                    |                    |
                    |  1. build_prompt() |  <-- reads ctx["text"]
                    |     prompt = "..." |
                    |                    |
                    |  2. engine.execute |  <-- sends prompt to LLM
                    |     raw = "Short.." |
                    |                    |
                    |  3. process_output |  <-- transforms raw output
                    |     result = "..." |
                    |                    |
                    |  4. Store result:  |
                    |     STRING -> ctx[skill_name] = result
                    |     DICT   -> ctx.merge(result)
                    |                    |
                    |  5. Record history |
                    +---------+----------+
                              |
                    +--------------------+
                    | "text": "Hello..." |
                    | "language": "FR"   |
                    | "summarize": "..." |  <-- new key added
                    +--------------------+
                              |
                    +---------v----------+
                    |  translate.run()   |
                    |                    |
                    |  reads ctx["summarize"]
                    |  writes ctx["translate"]
                    +---------+----------+
                              |
                    +--------------------+
                    | "text": "Hello..." |
                    | "language": "FR"   |
                    | "summarize": "..." |
                    | "translate": "..." |  <-- another key added
                    +--------------------+

Return value convention:

• String return -> stored as ctx[skill_name] (e.g., skill named "summarize" stores in ctx["summarize"])
• Dict return -> merged into context (e.g., return {"summary": "...", "keywords": [...]} adds both keys)

Accessing previous skill outputs:

@skill(name="step-two", model="claude-sonnet-4-6")
async def step_two(ctx):
    # Read output from a skill that returned a string
    previous_output = ctx["step-one"]      # key = skill name

    # Read output from a skill that returned a dict
    specific_field = ctx["summary"]        # key = dict key

    # Check if a key exists
    optional = ctx.get("maybe-missing", "default value")

    return f"Process: {previous_output}"

Context in different patterns:

Pattern	Context behavior
Sequential (>>)	Same context flows A -> B -> C. Each skill sees all previous outputs.
Parallel	Each parallel skill gets an isolated snapshot. Results merge back by named key.
Conditional	The selected route skill gets the full context.
MapReduce	Mapper gets a snapshot + ctx["item"] per item. Reducer gets ctx["results"] (list of mapper outputs).
Loop	Same context across iterations. Skill keeps overwriting the same keys.

Inspecting execution history:

result = await chain.run({"text": "Hello"})

# All intermediate results by skill name
result.results["summarize"]    # output of summarize skill
result.results["translate"]    # output of translate skill

# Full execution history: list of (skill_name, input_snapshot, output)
for name, input_snap, output in result.history:
    print(f"{name}: {output}")

Progressive Disclosure

SkillChain follows the agentskills.io progressive disclosure model. Skills loaded from directories go through three stages, loading more detail only when needed:

Stage 1: DISCOVERED          Stage 2: ACTIVATED           Stage 3: RESOURCES_LOADED
~100 tokens                  < 5000 tokens                Variable

+------------------+         +------------------+         +------------------+
| name             |         | name             |         | name             |
| description      |         | description      |         | description      |
|                  |  -----> | instructions     |  -----> | instructions     |
|  (nothing else)  | on first| metadata         | on first| metadata         |
|                  |  run()  | model            |  run()  | model            |
+------------------+         +------------------+         | build_prompt()   |
                                                          | process_output() |
                                                          +------------------+
     Skill.from_directory()       _ensure_activated()        _ensure_resources_loaded()
     SkillRegistry.scan()         automatic in run()         automatic in run()

This means you can register hundreds of skills in a registry and only pay the token cost for the ones that actually execute.

# Stage 1 only — just name + description loaded
s = Skill.from_directory("./my-skill/")
s.disclosure_stage   # DisclosureStage.DISCOVERED
s.instructions       # "" (not loaded yet)

# Stages 2+3 happen automatically on first run
await s.run({"input": "hello"})
s.disclosure_stage   # DisclosureStage.RESOURCES_LOADED
s.instructions       # "Full SKILL.md body here..."

Package Structure

skillchain/
├── __init__.py                 # Public API: Skill, skill, Chain, Parallel, etc.
├── exceptions.py               # SkillError, ChainError, ModelError, etc.
├── core/
│   ├── skill.py                # Skill base class, @skill decorator, from_directory()
│   ├── context.py              # SkillContext (shared data bus)
│   ├── chain.py                # Chain class (>> operator)
│   ├── engine.py               # ExecutionEngine (runs prompts through LLM)
│   └── disclosure.py           # ProgressiveLoader (3-stage lazy loading)
├── patterns/
│   ├── parallel.py             # Parallel(**named_skills)
│   ├── conditional.py          # Conditional(condition, routes, default)
│   ├── map_reduce.py           # MapReduce(mapper, reducer, input_key)
│   └── loop.py                 # Loop(skill, until, max_iterations)
├── models/
│   └── provider.py             # LiteLLM wrapper (ModelProvider)
├── registry/
│   ├── registry.py             # SkillRegistry (discover + load skills)
│   ├── parser.py               # SKILL.md YAML frontmatter parser
│   └── loaders/
│       ├── local.py            # Load from local directories
│       ├── url.py              # Load from URLs
│       └── package.py          # Load from pip packages
├── skills/                     # Built-in skills (agentskills.io format)
│   ├── read-file/SKILL.md
│   ├── write-file/SKILL.md
│   ├── list-files/SKILL.md
│   ├── summarize/SKILL.md
│   ├── extract-json/SKILL.md
│   └── classify/SKILL.md
└── telemetry/
    ├── tracing.py              # SkillTracer (OTel GenAI spans)
    └── attributes.py           # gen_ai.* attribute constants

---

Orchestration Patterns

All five patterns implement the Skill interface, so they compose with each other and with the >> operator.

1. Sequential (>>)

Skills execute one after another. Each skill sees all previous outputs in the context.

chain = skill_a >> skill_b >> skill_c
result = await chain.run({"input": "data"})

Context flow:

ctx --> [skill_a] --> ctx' --> [skill_b] --> ctx'' --> [skill_c] --> ctx'''

2. Parallel

Skills execute concurrently. Each gets an isolated snapshot of the context. Results merge back with named keys.

from skillchain import Parallel

analysis = Parallel(
    sentiment=analyze_sentiment,    # result -> ctx["sentiment"]
    entities=extract_entities,      # result -> ctx["entities"]
    keywords=extract_keywords,      # result -> ctx["keywords"]
)
result = await analysis.run({"text": "..."})

Context flow:

           +---> [sentiment] ---+
           |   (isolated copy)  |
ctx -------+---> [entities]  ---+----> ctx (merged results)
           |   (isolated copy)  |
           +---> [keywords]  ---+
              (isolated copy)

The named keys (sentiment=, entities=, keywords=) determine where each skill's output is stored in the merged context.

3. Conditional

Routes to different skills based on a condition function that inspects the context.

from skillchain import Conditional

router = Conditional(
    condition=lambda ctx: ctx["language"],   # returns route key
    routes={
        "python": review_python,
        "javascript": review_javascript,
    },
    default=review_generic,                  # fallback
)

Context flow:

ctx --> condition(ctx) = "python" --> [review_python] --> ctx'
                         "javascript" --> [review_javascript]
                         (other) --> [review_generic]

4. MapReduce

Fan out a mapper skill across a list of items (in parallel), then reduce all results.

from skillchain import MapReduce

batch_summarize = MapReduce(
    mapper=summarize_chunk,      # runs once per item
    reducer=combine_summaries,   # merges all mapper outputs
    input_key="chunks",          # ctx key containing the list
)

Context flow:

ctx["chunks"] = [item1, item2, item3]

          +---> [mapper](item=item1) ---> result1 ---+
          |                                           |
ctx ------+---> [mapper](item=item2) ---> result2 ---+---> ctx["results"] = [r1, r2, r3]
          |                                           |         |
          +---> [mapper](item=item3) ---> result3 ---+         v
                                                          [reducer](results) --> ctx'

Each mapper invocation receives a context snapshot with ctx["item"] set to the current item. The reducer receives all mapper outputs in ctx["results"].

5. Loop

Repeat a skill until a condition is met or max iterations reached.

from skillchain import Loop

refine = Loop(
    skill=improve_draft,
    until=lambda ctx: ctx.get("quality_score", 0) > 0.9,
    max_iterations=5,
)

Context flow:

ctx --> [improve_draft] --> check until() --> false --> [improve_draft] --> check until()
                                                                              |
                                                              true (or max) --+--> ctx
                                                                              |
                                                              ctx["loop_iterations"] = N

Composing Patterns

Every pattern is a Skill, so patterns nest inside each other:

# Parallel inside a chain
chain = fetch_data >> Parallel(
    analysis=analyze,
    summary=summarize,
) >> generate_report

# Loop containing a chain
refinement = Loop(
    skill=critique >> improve,          # chain inside loop
    until=lambda ctx: ctx.get("score", 0) > 0.9,
    max_iterations=3,
)

# MapReduce inside a chain with Conditional
pipeline = split_text >> MapReduce(
    mapper=Conditional(                 # conditional inside mapreduce
        condition=lambda ctx: ctx.get("type", "text"),
        routes={"code": analyze_code, "text": analyze_text},
    ),
    reducer=merge_results,
    input_key="chunks",
) >> format_output

---

Defining Skills

Decorator Style

The simplest way — for inline skills with straightforward logic:

from skillchain import skill

@skill(name="summarize", description="Summarize text", model="claude-sonnet-4-6")
async def summarize(ctx):
    # Return a string: sent to the LLM as the prompt
    # LLM response is stored in ctx["summarize"] (the skill name)
    return f"Summarize this:\n\n{ctx['text']}"

@skill(name="extract-data", description="Extract fields", model="claude-sonnet-4-6")
async def extract_data(ctx):
    # Return a dict: merged directly into context (no LLM call for the merge)
    return {"title": "extracted", "date": "2026-01-01"}

@skill(name="uppercase", description="Uppercase locally", model=None)
async def uppercase(ctx):
    # model=None: no LLM call, the return value IS the output
    return {"upper_text": ctx["text"].upper()}

Class Style

For skills with custom pre/post processing:

from skillchain import Skill, SkillContext

class SummarizeSkill(Skill):
    name = "summarize"
    description = "Summarize text concisely"
    model = "claude-sonnet-4-6"

    async def build_prompt(self, ctx: SkillContext) -> str:
        return f"Summarize in {ctx.get('max_sentences', 3)} sentences:\n\n{ctx['text']}"

    async def process_output(self, raw: str, ctx: SkillContext) -> dict:
        # Custom post-processing — return dict to merge into context
        return {
            "summary": raw,
            "summary_length": len(raw),
        }

summarize = SummarizeSkill()
chain = summarize >> next_skill

From SKILL.md Directory (agentskills.io)

Load skills from agentskills.io-compliant directories:

my-skill/
├── SKILL.md              # Frontmatter + instructions
└── scripts/
    └── handler.py        # build_prompt() + process_output()

<!-- my-skill/SKILL.md -->
---
name: my-skill
description: Does something useful. Use when the user needs X.
metadata:
  model: claude-sonnet-4-6
---

You are a helpful skill. Follow these instructions carefully.

## Steps
1. Read the input
2. Process it
3. Return the result

# my-skill/scripts/handler.py
async def build_prompt(ctx):
    return f"Process this: {ctx['input']}"

async def process_output(raw, ctx):
    return {"result": raw}

from skillchain import Skill

my_skill = Skill.from_directory("./my-skill/")
result = my_skill.run_sync({"input": "hello"})

---

Built-in Skills

SkillChain ships with six built-in skills, each as a proper agentskills.io directory:

Skill	Model	Reads	Writes
read_file	None (local)	ctx["file_path"]	ctx["content"]
write_file	None (local)	ctx["output_path"], ctx["content"]	ctx["write_status"]
list_files	None (local)	ctx["directory"], optional ctx["pattern"]	ctx["files"]
summarize	claude-sonnet-4-6	ctx["text"]	ctx["summary"]
extract_json	claude-sonnet-4-6	ctx["text"], optional ctx["schema"]	ctx["extracted"]
classify	claude-sonnet-4-6	ctx["text"], ctx["categories"]	ctx["classification"]

from skillchain.skills import read_file, write_file, summarize

# Chain built-in skills with custom skills
pipeline = read_file >> summarize >> write_file
pipeline.run_sync({
    "file_path": "article.txt",
    "output_path": "summary.txt",
})

---

Skill Registry

Discover and load skills from multiple sources:

from skillchain import SkillRegistry

registry = SkillRegistry()

# Local directories (scans for SKILL.md files)
registry.register_directory("./my-skills/")

# Remote URL
await registry.register_url("https://example.com/skills/my-skill/")

# pip-installed packages (Python entry points)
registry.discover_packages()

# Use a skill
skill = registry.get("data-analysis")
chain = skill >> format_output

# List all registered skills
print(registry.list())  # ["data-analysis", "other-skill", ...]

Skills in the registry stay in Stage 1 (discovery) — only name + description loaded. Full instructions and scripts load lazily when the skill actually runs.

---

Per-Skill Model Selection

Each skill can use a different LLM via LiteLLM model strings:

# Fast + cheap for simple tasks
@skill(name="classify", model="claude-haiku-4-5-20251001")
async def classify(ctx): ...

# Powerful for complex reasoning
@skill(name="analyze", model="claude-sonnet-4-6")
async def analyze(ctx): ...

# Use OpenAI for a specific skill
@skill(name="generate-image-prompt", model="gpt-4o")
async def generate_image_prompt(ctx): ...

# Override model for an entire chain
chain = (classify >> analyze).with_default_model("claude-sonnet-4-6")

Set API keys via environment variables:

export ANTHROPIC_API_KEY="..."    # Claude models
export OPENAI_API_KEY="..."       # GPT models
export GEMINI_API_KEY="..."       # Gemini models

---

OpenTelemetry Tracing

SkillChain integrates OpenTelemetry with GenAI semantic conventions for full observability of skill orchestration.

pip install skillchain[telemetry]

Enable tracing

from skillchain.telemetry import SkillTracer
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor, ConsoleSpanExporter

# Configure with console output (or any OTel exporter: Jaeger, OTLP, etc.)
provider = TracerProvider()
provider.add_span_processor(SimpleSpanProcessor(ConsoleSpanExporter()))
SkillTracer.configure(provider=provider)

# All skill executions now produce spans automatically
chain = summarize >> translate
await chain.run({"text": "..."})   # spans emitted!

Span hierarchy

For a chain summarize >> Parallel(topics, sentiment) >> report:

invoke_workflow "chain-summarize-parallel-..." (sequential)
├── invoke_agent "summarize"
│   └── chat "claude-sonnet-4-6" (CLIENT, provider=anthropic)
├── invoke_workflow "parallel-sentiment-topics" (parallel)
│   ├── invoke_agent "sentiment"
│   │   └── chat "claude-haiku-4-5" (CLIENT, provider=anthropic)
│   └── invoke_agent "topics"
│       └── chat "claude-haiku-4-5" (CLIENT, provider=anthropic)
└── invoke_agent "report"
    └── chat "claude-sonnet-4-6" (CLIENT, provider=anthropic)

Span types and attributes

Span	OTel Operation	Kind	Key Attributes
Skill execution	invoke_agent	INTERNAL	gen_ai.agent.name, gen_ai.request.model
Chain/Pattern	invoke_workflow	INTERNAL	gen_ai.workflow.name, skillchain.pattern.type
LLM call	chat	CLIENT	gen_ai.request.model, gen_ai.provider.name
Local skill	execute_tool	INTERNAL	gen_ai.tool.name, gen_ai.tool.type

Tracing is no-op when OTel is not installed — no performance impact.

---

Examples

Example 1: Simple Chain with Claude

from skillchain import skill

@skill(name="summarize", model="claude-sonnet-4-6")
async def summarize(ctx):
    return f"Summarize this text in 2 sentences:\n\n{ctx['text']}"

@skill(name="translate", model="claude-haiku-4-5-20251001")
async def translate(ctx):
    language = ctx.get("language", "French")
    return f"Translate to {language}:\n\n{ctx['summarize']}"

chain = summarize >> translate
result = chain.run_sync({
    "text": "SkillChain is a Python SDK for orchestrating AI skills...",
    "language": "Spanish",
})
print(result["summarize"])    # English summary
print(result["translate"])    # Spanish translation

Example 2: Parallel Analysis

Run multiple analyses concurrently, then combine:

from skillchain import skill, Parallel

@skill(name="sentiment", model="claude-haiku-4-5-20251001")
async def analyze_sentiment(ctx):
    return f"Sentiment (one word: positive/negative/neutral):\n\n{ctx['text']}"

@skill(name="entities", model="claude-haiku-4-5-20251001")
async def extract_entities(ctx):
    return f"Extract named entities as comma-separated list:\n\n{ctx['text']}"

@skill(name="report", model="claude-sonnet-4-6")
async def generate_report(ctx):
    return (
        f"Analysis report given:\n"
        f"Sentiment: {ctx['sentiment']}\n"
        f"Entities: {ctx['entities']}\n"
        f"Text: {ctx['text']}"
    )

chain = Parallel(sentiment=analyze_sentiment, entities=extract_entities) >> generate_report
result = chain.run_sync({"text": "Apple CEO Tim Cook announced..."})

Example 3: Document Pipeline with Built-in Skills

from skillchain import skill, MapReduce
from skillchain.skills import read_file, write_file

@skill(name="chunk-text", description="Split into chunks", model=None)
async def chunk_text(ctx):
    text = ctx["content"]
    size = ctx.get("chunk_size", 2000)
    return {"chunks": [text[i:i+size] for i in range(0, len(text), size)]}

@skill(name="summarize-chunk", model="claude-sonnet-4-6")
async def summarize_chunk(ctx):
    return f"Summarize:\n\n{ctx['item']}"

@skill(name="combine", model="claude-sonnet-4-6")
async def combine(ctx):
    parts = "\n---\n".join(r.get("summarize-chunk", str(r)) for r in ctx["results"])
    return f"Combine into one summary:\n\n{parts}"

pipeline = read_file >> chunk_text >> MapReduce(
    mapper=summarize_chunk,
    reducer=combine,
    input_key="chunks",
) >> write_file

pipeline.run_sync({
    "file_path": "long_document.txt",
    "output_path": "summary.txt",
})

Example 4: Conditional Routing

from skillchain import skill, Conditional

@skill(name="detect-lang", model="claude-haiku-4-5-20251001")
async def detect_language(ctx):
    return f"What language is this code? Reply: python/javascript/other\n\n{ctx['code']}"

@skill(name="review-python", model="claude-sonnet-4-6")
async def review_python(ctx):
    return f"Review this Python code:\n\n{ctx['code']}"

@skill(name="review-js", model="claude-sonnet-4-6")
async def review_js(ctx):
    return f"Review this JavaScript code:\n\n{ctx['code']}"

@skill(name="review-generic", model="claude-sonnet-4-6")
async def review_generic(ctx):
    return f"Review this code:\n\n{ctx['code']}"

chain = detect_language >> Conditional(
    condition=lambda ctx: ctx["detect-lang"].strip().lower(),
    routes={"python": review_python, "javascript": review_js},
    default=review_generic,
)
result = chain.run_sync({"code": "def hello(): print('hi')"})

Example 5: All Five Patterns Combined

No API key needed — runs entirely locally:

from skillchain import skill, Parallel, Conditional, MapReduce, Loop

@skill(name="generate-data", model=None)
async def generate_data(ctx):
    return {"items": ["apple", "banana", "cherry", "date", "elderberry"]}

@skill(name="count-chars", model=None)
async def count_chars(ctx):
    return {"counted": {"word": ctx["item"], "length": len(ctx["item"])}}

@skill(name="find-longest", model=None)
async def find_longest(ctx):
    longest = max(ctx["results"], key=lambda r: r["counted"]["length"])
    return {"longest": longest["counted"]["word"]}

@skill(name="categorize", model=None)
async def categorize(ctx):
    return {"category": "long" if len(ctx["longest"]) > 5 else "short"}

@skill(name="handle-long", model=None)
async def handle_long(ctx):
    return {"verdict": f"'{ctx['longest']}' is long ({len(ctx['longest'])} chars)"}

@skill(name="handle-short", model=None)
async def handle_short(ctx):
    return {"verdict": f"'{ctx['longest']}' is short ({len(ctx['longest'])} chars)"}

@skill(name="get-upper", model=None)
async def get_upper(ctx):
    return {"upper": ctx["longest"].upper()}

@skill(name="get-reverse", model=None)
async def get_reverse(ctx):
    return {"reverse": ctx["longest"][::-1]}

@skill(name="add-emphasis", model=None)
async def add_emphasis(ctx):
    ctx["emphasis_count"] = ctx.get("emphasis_count", 0) + 1
    return {"emphasized": ctx.get("emphasized", ctx["verdict"]) + "!"}

# Compose all 5 patterns:
chain = (
    generate_data                                          # Sequential
    >> MapReduce(mapper=count_chars,                        # MapReduce
                 reducer=find_longest, input_key="items")
    >> categorize
    >> Parallel(upper=get_upper, reverse=get_reverse)      # Parallel
    >> Conditional(condition=lambda ctx: ctx["category"],   # Conditional
                   routes={"long": handle_long, "short": handle_short})
    >> Loop(skill=add_emphasis,                             # Loop
            until=lambda ctx: ctx.get("emphasis_count", 0) >= 3,
            max_iterations=5)
)

result = chain.run_sync({})
print(result["longest"])      # "elderberry"
print(result["upper"])        # "ELDERBERRY"
print(result["reverse"])      # "yrrebredle"
print(result["emphasized"])   # "'elderberry' is long (10 chars)!!!"

Example 6: Custom Skills from Directories (agentskills.io)

Create your own skills as directories with SKILL.md files, then load and orchestrate them. This is the agentskills.io-native way to build reusable, portable skills.

Directory structure:

my-project/
├── main.py
└── skills/
    ├── extract-keywords/
    │   ├── SKILL.md
    │   └── scripts/handler.py
    ├── generate-hashtags/
    │   ├── SKILL.md
    │   └── scripts/handler.py
    └── format-post/
        ├── SKILL.md
        └── scripts/handler.py

Skill 1 — skills/extract-keywords/SKILL.md:

---
name: extract-keywords
description: Extract keywords from text. Reads ctx['text'], writes ctx['keywords'] as a list.
metadata:
  model: claude-haiku-4-5-20251001
---

You are a keyword extraction specialist. Given any text, identify the most important keywords and phrases.

## Instructions

1. Read the input text carefully
2. Identify 5-8 key terms that capture the main topics
3. Return them as a comma-separated list
4. Focus on nouns and noun phrases, not generic words

Skill 1 — skills/extract-keywords/scripts/handler.py:

async def build_prompt(ctx):
    return (
        "Extract 5-8 important keywords from this text. "
        "Return only a comma-separated list, nothing else.\n\n"
        f"{ctx['text']}"
    )


async def process_output(raw, ctx):
    keywords = [k.strip() for k in raw.split(",")]
    return {"keywords": keywords}

Skill 2 — skills/generate-hashtags/SKILL.md:

---
name: generate-hashtags
description: Generate social media hashtags from keywords. Reads ctx['keywords'], writes ctx['hashtags'].
metadata:
  model: claude-haiku-4-5-20251001
---

You are a social media expert. Convert keywords into engaging hashtags suitable for LinkedIn and Twitter.

## Instructions

1. Take the provided keywords
2. Transform each into a hashtag (camelCase, no spaces)
3. Add 2-3 trending/general hashtags relevant to the topic
4. Return as a space-separated list of hashtags

Skill 2 — skills/generate-hashtags/scripts/handler.py:

async def build_prompt(ctx):
    keywords = ", ".join(ctx["keywords"])
    return (
        "Convert these keywords into social media hashtags. "
        "Use camelCase, add 2-3 general trending hashtags. "
        "Return only the hashtags as a space-separated list.\n\n"
        f"Keywords: {keywords}"
    )


async def process_output(raw, ctx):
    hashtags = [h.strip() for h in raw.split() if h.startswith("#")]
    if not hashtags:
        hashtags = [f"#{h.strip()}" for h in raw.split() if h.strip()]
    return {"hashtags": hashtags}

Skill 3 — skills/format-post/SKILL.md:

---
name: format-post
description: Format a social media post from text, keywords, and hashtags. Reads ctx['text'], ctx['keywords'], ctx['hashtags'], writes ctx['post'].
metadata:
  model: claude-sonnet-4-6
---

You are a content writer. Create an engaging social media post that summarizes the original content and incorporates the provided hashtags.

## Instructions

1. Read the original text for context
2. Write a concise, engaging post (2-3 sentences max)
3. Append the hashtags at the end
4. Make it suitable for LinkedIn

Skill 3 — skills/format-post/scripts/handler.py:

async def build_prompt(ctx):
    keywords = ", ".join(ctx["keywords"])
    hashtags = " ".join(ctx["hashtags"])
    return (
        "Write a short, engaging LinkedIn post (2-3 sentences) based on this content. "
        "End with the hashtags on a new line.\n\n"
        f"Original text: {ctx['text']}\n\n"
        f"Keywords: {keywords}\n\n"
        f"Hashtags to include: {hashtags}"
    )


async def process_output(raw, ctx):
    return {"post": raw}

Orchestrator — main.py:

import os
import sys
from pathlib import Path
from skillchain import Skill, SkillRegistry

skills_dir = Path(__file__).parent / "skills"

# --- Option A: Load individual skills by path ---

extract_keywords = Skill.from_directory(str(skills_dir / "extract-keywords"))
generate_hashtags = Skill.from_directory(str(skills_dir / "generate-hashtags"))
format_post = Skill.from_directory(str(skills_dir / "format-post"))

# At this point, only name + description are loaded (Stage 1: DISCOVERED)
print(extract_keywords.disclosure_stage)  # DisclosureStage.DISCOVERED

# Chain them
chain = extract_keywords >> generate_hashtags >> format_post

result = chain.run_sync({
    "text": "SkillChain is a new Python SDK that lets developers orchestrate AI skills "
            "in composable patterns. It supports sequential, parallel, conditional, "
            "map-reduce, and loop patterns with per-skill model selection."
})

# After run, all stages completed (Stage 3: RESOURCES_LOADED)
print(f"Keywords: {result['keywords']}")
print(f"Hashtags: {' '.join(result['hashtags'])}")
print(f"Post:\n{result['post']}")


# --- Option B: Use SkillRegistry to scan a directory ---

registry = SkillRegistry()
registry.register_directory(str(skills_dir))

print(registry.list())  # ['extract-keywords', 'generate-hashtags', 'format-post']

kw = registry.get("extract-keywords")
ht = registry.get("generate-hashtags")
fp = registry.get("format-post")

chain2 = kw >> ht >> fp
result2 = chain2.run_sync({"text": "OpenTelemetry provides observability for distributed systems."})
print(result2["post"])

Run:

export ANTHROPIC_API_KEY="your-key-here"
python main.py

---

Error Handling

from skillchain.exceptions import (
    SkillError,            # Base for all SkillChain errors
    SkillNotFoundError,    # Skill not in registry
    SkillExecutionError,   # Skill failed (wraps original error + context snapshot)
    SkillValidationError,  # Invalid SKILL.md
    ModelError,            # LLM call failed (wraps LiteLLM error)
    ChainError,            # Chain failed (includes position in chain)
)

try:
    result = await chain.run({"text": "..."})
except SkillExecutionError as e:
    print(f"Skill '{e.skill_name}' failed: {e.original_error}")
    print(f"Context at failure: {e.context_snapshot}")
except ChainError as e:
    print(f"Chain failed at skill '{e.skill_name}' (position {e.position})")

With OTel tracing enabled, errors are automatically recorded on spans with StatusCode.ERROR, exception details, and full stack traces.

---

License

MIT