OpenSkills 入门：用 SKILL.md 定义可控的 AI Agent 技能调用

weekly-report-to-annual/
├── SKILL.md
├── scripts/
│   ├── fetch_emails.py
│   └── save_report.py
├── references/
│   └── annual-review-template.md
└── demo.py

---
name: weekly-report-to-annual
description: 从周报邮件生成年度复盘报告
version: 1.0.0

triggers:
  - 年度总结
  - 周报汇总
  - annual review

scripts:
  - name: fetch_emails
    path: scripts/fetch_emails.py
    description: 从 IMAP 服务器获取邮件

  - name: save_report
    path: scripts/save_report.py
    description: 保存报告到本地

references:
  - path: references/annual-review-template.md
    mode: always
    description: 年度复盘报告模板
---

# 周报年度复盘技能

你是一个年度复盘助手，需要根据周报内容生成结构化报告。

## 执行步骤

1. 如果用户提供了邮箱配置，调用 [INVOKE:fetch_emails] 获取周报。
2. 分析周报内容，提取以下信息：
   - 重要项目和成果
   - 技术成长和学习
   - 遇到的挑战和解决方案
   - 团队协作和贡献
3. 按照 references 中的年度复盘模板生成报告。
4. 调用 [INVOKE:save_report] 保存到本地。

import asyncio
from openskills import create_agent

async def main():
    agent = await create_agent(
        skill_paths=["./examples/weekly-report-to-annual"],
        api_key="your-api-key",
        model="gpt-4",
    )

    response = await agent.chat("""
    帮我生成张三的 2025 年度复盘报告。
    从飞书邮箱获取所有标题包含“周报｜张三”的邮件。
    """)

    print(response.content)

asyncio.run(main())

# 张三-2025 年度复盘报告

## 1. 目标复盘

| 年度目标 | 达成情况 | 后续行动 |
|---|---|---|
| API 网关性能优化 | QPS 从 2000 提高到 8000 | 继续压测并优化连接池 |
| 订单系统微服务化 | 拆分出用户、订单、支付、库存 4 个服务 | 扩展到更多业务模块 |
| 云原生能力建设 | Kubernetes 落地到生产环境 | 深入服务网格和可观测性 |

## 2. 工作亮点

- API 网关重构：通过连接池优化和对象复用，把 QPS（每秒查询数）从 2000 提高到 8000。
- 订单系统微服务化：部署时间从 30 分钟缩短到 5 分钟。
- 支付系统重构：支付成功率从 95% 提高到 99.2%。

## 3. 经验与教训

1. 性能优化要先定位瓶颈，再做针对性修改。
2. 微服务拆分应按业务边界划分，而不是按技术层次拆分。
3. 第三方支付接口需要提前设计超时、重试和降级机制。

#!/usr/bin/env python3
"""从飞书邮箱获取周报邮件"""

import imaplib
import json
import sys
from concurrent.futures import ThreadPoolExecutor, as_completed

IMAP_HOST = "imap.feishu.cn"
IMAP_PORT = 993

def list_folders(email: str, password: str) -> list[str]:
    mail = imaplib.IMAP4_SSL(IMAP_HOST, IMAP_PORT)
    mail.login(email, password)

    status, data = mail.list()
    mail.logout()

    if status != "OK":
        return []

    folders = []
    for item in data:
        line = item.decode("utf-8", errors="ignore")
        folder = line.split(' "/" ')[-1].strip('"')
        folders.append(folder)

    return folders

def search_folder_with_own_connection(
    folder: str,
    email: str,
    password: str,
    keyword: str,
    year: int,
) -> tuple[str, list[dict]]:
    mail = imaplib.IMAP4_SSL(IMAP_HOST, IMAP_PORT)
    mail.login(email, password)

    try:
        mail.select(folder, readonly=True)

        # 示例搜索条件，真实场景可按邮件服务器能力调整
        status, message_ids = mail.search(None, "ALL")
        if status != "OK":
            return folder, []

        results = []

        for msg_id in message_ids[0].split():
            status, msg_data = mail.fetch(msg_id, "(RFC822)")
            if status != "OK":
                continue

            # 这里省略邮件解析逻辑
            # 可解析 Subject、Date、From、Body 等字段
            parsed = parse_email(msg_data)

            if keyword in parsed["subject"] and str(year) in parsed["date"]:
                results.append(parsed)

        return folder, results

    finally:
        mail.logout()

def parse_email(msg_data) -> dict:
    """示例占位函数：真实实现需要解析 MIME 邮件内容"""
    return {
        "subject": "",
        "date": "",
        "body": "",
    }

def fetch_emails_from_feishu(
    email: str,
    password: str,
    keyword: str,
    year: int,
) -> list[dict]:
    folders = list_folders(email, password)
    all_emails = []

    with ThreadPoolExecutor(max_workers=5) as executor:
        futures = {
            executor.submit(
                search_folder_with_own_connection,
                folder,
                email,
                password,
                keyword,
                year,
            ): folder
            for folder in folders
        }

        for future in as_completed(futures):
            folder, emails = future.result()
            all_emails.extend(emails)

    return all_emails

if __name__ == "__main__":
    input_data = json.loads(sys.stdin.read())

    emails = fetch_emails_from_feishu(
        email=input_data["email"],
        password=input_data["password"],
        keyword=input_data.get("keyword", "周报｜张三"),
        year=input_data.get("year", 2025),
    )

    print(json.dumps({
        "total": len(emails),
        "emails": emails,
    }, ensure_ascii=False))

openskills/
├── core/
│   ├── manager.py        # SkillManager：技能管理
│   ├── matcher.py        # 智能匹配
│   ├── parser.py         # SKILL.md 解析
│   └── executor.py       # 脚本执行
│
├── models/
│   ├── metadata.py       # Layer 1：元数据
│   ├── instruction.py    # Layer 2：指令
│   └── resource.py       # Layer 3：资源
│
├── llm/
│   ├── openai_compat.py  # OpenAI 兼容接口
│   └── prompt_builder.py # 提示词构建
│
├── sandbox/
│   ├── client.py         # 沙箱 HTTP 客户端
│   └── executor.py       # 沙箱执行器
│
└── agent.py              # SkillAgent 高级 API

import asyncio
from pathlib import Path
from openskills import SkillManager

async def main():
    manager = SkillManager([Path("./skills")])

    # 发现技能，只加载 Metadata
    await manager.discover()

    # 根据用户输入匹配技能
    skills = manager.match("生成年度报告")
    skill = skills[0]

    # 手动加载 Instruction
    await manager.load_instruction(skill.name)

    # 手动加载 References
    await manager.load_references(skill.name)

    # 手动执行脚本
    result = await manager.execute_script(
        skill.name,
        "fetch_emails",
        {
            "email": "user@example.com",
            "password": "use-secret-manager",
            "keyword": "周报｜张三",
            "year": 2025,
        },
    )

    print(result)

asyncio.run(main())

import asyncio
from openskills import create_agent

async def main():
    agent = await create_agent(
        skill_paths=["./skills"],
        api_key="your-key",
        auto_select_skill=True,
        auto_execute_scripts=True,
        use_sandbox=True,
    )

    response = await agent.chat("帮我生成年度报告")

    print(response.content)
    print(f"使用的技能: {response.skill_used}")

asyncio.run(main())

docker run --rm -p 8080:8080 ghcr.io/agent-infra/sandbox:latest

agent = await create_agent(
    skill_paths=["./skills"],
    api_key="your-key",
    use_sandbox=True,
    sandbox_base_url="http://localhost:8080",
    auto_execute_scripts=True,
)

dependency:
  python:
    - pandas>=2.0.0
    - requests>=2.31.0
  npm:
    - mermaid@latest
  system:
    - apt-get install -y imagemagick

references:
  - path: references/guidelines.md
    mode: always
    description: 必读规范

  - path: references/financial-handbook.md
    mode: explicit
    condition: 当讨论涉及财务、预算、成本时

  - path: references/examples.md
    mode: implicit
    description: 示例参考

my-skill/
├── SKILL.md
└── references/
    ├── template.md
    ├── examples.md
    └── appendix/
        └── glossary.md

from openskills.llm import OpenAICompatClient

# OpenAI
client = OpenAICompatClient(
    api_key="sk-...",
    model="gpt-4",
)

# Azure OpenAI
client = OpenAICompatClient(
    api_key="...",
    base_url="https://xxx.openai.azure.com/",
    model="gpt-4",
)

# Ollama 本地模型
client = OpenAICompatClient(
    base_url="http://localhost:11434/v1",
    model="llama3.2",
    api_key="ollama",
)

# 列出所有技能
openskills list

# 查看技能详情
openskills show weekly-report-to-annual

# 验证技能格式
openskills validate ./my-skill/

# 测试匹配
openskills match "生成年度报告"

mkdir my-translator
cd my-translator

---
name: smart-translator
description: 智能翻译，自动识别源语言和目标语言
version: 1.0.0

triggers:
  - 翻译
  - translate
  - 翻成
---

# 智能翻译技能

你是一个专业的翻译助手。

## 翻译原则

1. 自动识别源语言和目标语言。
2. 保持原文的语气和风格。
3. 避免生硬的机翻腔。
4. 技术术语需要准确。

## 执行步骤

1. 识别源语言。
2. 识别目标语言；如果用户未指定，默认在中文和英文之间互译。
3. 翻译内容。
4. 检查术语、语气和格式。

import asyncio
from openskills import create_agent

async def main():
    agent = await create_agent(
        skill_paths=["./my-translator"],
        api_key="your-key",
    )

    response = await agent.chat("""
    翻译成英文：
    这个项目让 AI Agent 的技能调用不再是黑盒。
    """)

    print(response.content)

asyncio.run(main())

This project makes AI Agent skill invocation transparent.