Back to Details

d6e-docker-stf-development

Compare original and translation side by side

🇺🇸

Original

English
🇨🇳

Translation

Chinese

D6E Docker STF Development

D6E Docker STF 开发指南

Overview

概述

Docker STFs are containerized applications that execute as workflow steps in D6E. They read JSON from stdin, process data with custom logic, access workspace databases via internal API, and output JSON to stdout.
Docker STF是作为D6E工作流步骤执行的容器化应用。它们从stdin读取JSON数据，通过自定义逻辑处理数据，通过内部API访问工作区数据库，并将JSON数据输出到stdout。

When to Use

适用场景

Apply this skill when users request:
  • "Create a D6E Docker STF that..."
  • "Build a custom STF for D6E that..."
  • "I need a Docker-based workflow step..."
  • "Help me create a data processing function for D6E"
当用户提出以下需求时，可使用本技能：
  • "创建一个D6E Docker STF，用于..."
  • "为D6E构建一个自定义STF，用于..."
  • "我需要一个基于Docker的工作流步骤..."
  • "帮我为D6E创建一个数据处理函数"

Core Concepts

核心概念

Input Format

输入格式

Docker STFs receive this JSON via stdin:
json
{
  "workspace_id": "UUID",
  "stf_id": "UUID",
  "caller": "UUID | null",
  "api_url": "http://api:8080",
  "api_token": "internal_token",
  "input": {
    "operation": "...",
    ...user-defined parameters
  },
  "sources": {
    "step_name": {
      "output": {...previous step data}
    }
  }
}
Docker STFs通过stdin接收如下JSON数据：
json
{
  "workspace_id": "UUID",
  "stf_id": "UUID",
  "caller": "UUID | null",
  "api_url": "http://api:8080",
  "api_token": "internal_token",
  "input": {
    "operation": "...",
    ...user-defined parameters
  },
  "sources": {
    "step_name": {
      "output": {...previous step data}
    }
  }
}

Output Format

输出格式

Success:
json
{
  "output": {
    "status": "success",
    ...custom result data
  }
}
Error:
json
{
  "error": "Error message",
  "type": "ErrorType"
}
成功响应：
json
{
  "output": {
    "status": "success",
    ...custom result data
  }
}
错误响应：
json
{
  "error": "Error message",
  "type": "ErrorType"
}

SQL API Access

SQL API 访问

Execute SQL via internal API:
Endpoint: 
POST /api/v1/workspaces/{workspace_id}/sql
Headers:
Authorization: Bearer {api_token}
X-Internal-Bypass: true
X-Workspace-ID: {workspace_id}
X-STF-ID: {stf_id}
Request:
json
{ "sql": "SELECT * FROM my_table LIMIT 10" }
Restrictions:
  • No DDL (CREATE, DROP, ALTER)
  • Policy-controlled access
  • Workspace scope only
通过内部API执行SQL：
端点： 
POST /api/v1/workspaces/{workspace_id}/sql
请求头：
Authorization: Bearer {api_token}
X-Internal-Bypass: true
X-Workspace-ID: {workspace_id}
X-STF-ID: {stf_id}
请求体：
json
{ "sql": "SELECT * FROM my_table LIMIT 10" }
限制：
  • 不支持DDL语句（CREATE、DROP、ALTER）
  • 受策略控制访问权限
  • 仅限定工作区范围

Quick Start

快速开始

Python Implementation

Python 实现

main.py:
python
#!/usr/bin/env python3
import sys
import json
import requests
import logging

logging.basicConfig(stream=sys.stderr, level=logging.INFO)

def execute_sql(api_url, api_token, workspace_id, stf_id, sql):
    """Execute SQL via D6E internal API"""
    url = f"{api_url}/api/v1/workspaces/{workspace_id}/sql"
    headers = {
        "Authorization": f"Bearer {api_token}",
        "X-Internal-Bypass": "true",
        "X-Workspace-ID": workspace_id,
        "X-STF-ID": stf_id,
        "Content-Type": "application/json"
    }
    response = requests.post(url, json={"sql": sql}, headers=headers)
    response.raise_for_status()
    return response.json()

def main():
    try:
        input_data = json.load(sys.stdin)
        user_input = input_data["input"]

        # Your business logic here
        result = {"status": "success", "message": "Processed"}

        print(json.dumps({"output": result}))
    except Exception as e:
        logging.error(f"Error: {str(e)}", exc_info=True)
        print(json.dumps({"error": str(e), "type": type(e).__name__}))
        sys.exit(1)

if __name__ == "__main__":
    main()
Dockerfile:
dockerfile
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY main.py .
RUN chmod +x main.py
ENTRYPOINT ["python3", "main.py"]
requirements.txt:
requests>=2.31.0
main.py：
python
#!/usr/bin/env python3
import sys
import json
import requests
import logging

logging.basicConfig(stream=sys.stderr, level=logging.INFO)

def execute_sql(api_url, api_token, workspace_id, stf_id, sql):
    """Execute SQL via D6E internal API"""
    url = f"{api_url}/api/v1/workspaces/{workspace_id}/sql"
    headers = {
        "Authorization": f"Bearer {api_token}",
        "X-Internal-Bypass": "true",
        "X-Workspace-ID": workspace_id,
        "X-STF-ID": stf_id,
        "Content-Type": "application/json"
    }
    response = requests.post(url, json={"sql": sql}, headers=headers)
    response.raise_for_status()
    return response.json()

def main():
    try:
        input_data = json.load(sys.stdin)
        user_input = input_data["input"]

        # Your business logic here
        result = {"status": "success", "message": "Processed"}

        print(json.dumps({"output": result}))
    except Exception as e:
        logging.error(f"Error: {str(e)}", exc_info=True)
        print(json.dumps({"error": str(e), "type": type(e).__name__}))
        sys.exit(1)

if __name__ == "__main__":
    main()
Dockerfile：
dockerfile
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY main.py .
RUN chmod +x main.py
ENTRYPOINT ["python3", "main.py"]
requirements.txt：
requests>=2.31.0

Node.js Implementation

Node.js 实现

index.js:
javascript
const axios = require("axios");

async function executeSql(apiUrl, apiToken, workspaceId, stfId, sql) {
  const response = await axios.post(
    `${apiUrl}/api/v1/workspaces/${workspaceId}/sql`,
    { sql },
    {
      headers: {
        Authorization: `Bearer ${apiToken}`,
        "X-Internal-Bypass": "true",
        "X-Workspace-ID": workspaceId,
        "X-STF-ID": stfId,
        "Content-Type": "application/json",
      },
    }
  );
  return response.data;
}

async function main() {
  try {
    const input = await readStdin();
    const data = JSON.parse(input);

    // Your business logic here
    const result = { status: "success", message: "Processed" };

    console.log(JSON.stringify({ output: result }));
  } catch (error) {
    console.error("Error:", error.message);
    console.log(
      JSON.stringify({
        error: error.message,
        type: error.name,
      })
    );
    process.exit(1);
  }
}

function readStdin() {
  return new Promise((resolve) => {
    let data = "";
    process.stdin.on("data", (chunk) => (data += chunk));
    process.stdin.on("end", () => resolve(data));
  });
}

main();
Dockerfile:
dockerfile
FROM node:18-slim
WORKDIR /app
COPY package*.json ./
RUN npm ci --omit=dev
COPY index.js .
ENTRYPOINT ["node", "index.js"]
index.js：
javascript
const axios = require("axios");

async function executeSql(apiUrl, apiToken, workspaceId, stfId, sql) {
  const response = await axios.post(
    `${apiUrl}/api/v1/workspaces/${workspaceId}/sql`,
    { sql },
    {
      headers: {
        Authorization: `Bearer ${apiToken}`,
        "X-Internal-Bypass": "true",
        "X-Workspace-ID": workspaceId,
        "X-STF-ID": stfId,
        "Content-Type": "application/json",
      },
    }
  );
  return response.data;
}

async function main() {
  try {
    const input = await readStdin();
    const data = JSON.parse(input);

    // Your business logic here
    const result = { status: "success", message: "Processed" };

    console.log(JSON.stringify({ output: result }));
  } catch (error) {
    console.error("Error:", error.message);
    console.log(
      JSON.stringify({
        error: error.message,
        type: error.name,
      })
    );
    process.exit(1);
  }
}

function readStdin() {
  return new Promise((resolve) => {
    let data = "";
    process.stdin.on("data", (chunk) => (data += chunk));
    process.stdin.on("end", () => resolve(data));
  });
}

main();
Dockerfile：
dockerfile
FROM node:18-slim
WORKDIR /app
COPY package*.json ./
RUN npm ci --omit=dev
COPY index.js .
ENTRYPOINT ["node", "index.js"]

Implementation Checklist

实现检查清单

When creating a Docker STF, ensure:
  • Reads JSON from stdin
  • Outputs JSON to stdout (
    {"output": {...}}
    )
  • Logs to stderr (stdout is for results only)
  • Handles errors gracefully
  • Uses small base images (e.g., 
    python:3.11-slim
    )
  • Includes error type in error responses
  • Validates input parameters
  • Uses environment variables for configuration
创建Docker STF时，请确保满足以下要求：
  • 从stdin读取JSON数据
  • 向stdout输出JSON数据（格式为
    {"output": {...}}
    ）
  • 向stderr输出日志（stdout仅用于返回结果）
  • 优雅处理错误
  • 使用轻量级基础镜像（例如
    python:3.11-slim
    ）
  • 错误响应中包含错误类型
  • 验证输入参数
  • 使用环境变量进行配置

Best Practices

最佳实践

Security

安全性

  • Never log sensitive data (tokens, passwords)
  • Validate all user inputs
  • Use parameterized SQL queries
  • Keep dependencies up-to-date
  • 切勿记录敏感数据（令牌、密码等）
  • 验证所有用户输入
  • 使用参数化SQL查询
  • 保持依赖为最新版本

Performance

性能

  • Use multi-stage builds to reduce image size
  • Minimize dependencies
  • Add 
    .dockerignore
    to exclude unnecessary files
  • Cache pip/npm installations
  • 使用多阶段构建来减小镜像大小
  • 最小化依赖数量
  • 添加
    .dockerignore
    文件排除不必要的文件
  • 缓存pip/npm安装过程

Error Handling

错误处理

python
try:
    # Your logic
    result = process_data(input_data)
    print(json.dumps({"output": result}))
except ValueError as e:
    # Validation errors
    logging.error(f"Validation error: {str(e)}")
    print(json.dumps({"error": str(e), "type": "ValidationError"}))
    sys.exit(1)
except Exception as e:
    # Unexpected errors
    logging.error(f"Unexpected error: {str(e)}", exc_info=True)
    print(json.dumps({"error": str(e), "type": type(e).__name__}))
    sys.exit(1)
python
try:
    # Your logic
    result = process_data(input_data)
    print(json.dumps({"output": result}))
except ValueError as e:
    # Validation errors
    logging.error(f"Validation error: {str(e)}")
    print(json.dumps({"error": str(e), "type": "ValidationError"}))
    sys.exit(1)
except Exception as e:
    # Unexpected errors
    logging.error(f"Unexpected error: {str(e)}", exc_info=True)
    print(json.dumps({"error": str(e), "type": type(e).__name__}))
    sys.exit(1)

Logging

日志记录

python
import logging
python
import logging

Log to stderr

Log to stderr

logging.basicConfig( stream=sys.stderr, level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s' )
logging.info("Processing started") logging.debug(f"Input: {input_data}") # Detailed logs logging.warning("Deprecated operation used") logging.error("Failed to process", exc_info=True)
undefined
logging.basicConfig( stream=sys.stderr, level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s' )
logging.info("Processing started") logging.debug(f"Input: {input_data}") # Detailed logs logging.warning("Deprecated operation used") logging.error("Failed to process", exc_info=True)
undefined

Common Patterns

常见模式

Data Validation Pattern

数据验证模式

python
def validate_input(user_input):
    required_fields = ["operation", "table_name"]
    for field in required_fields:
        if field not in user_input:
            raise ValueError(f"Missing required field: {field}")

    if user_input["operation"] not in ["query", "insert", "update"]:
        raise ValueError(f"Invalid operation: {user_input['operation']}")

    return True
python
def validate_input(user_input):
    required_fields = ["operation", "table_name"]
    for field in required_fields:
        if field not in user_input:
            raise ValueError(f"Missing required field: {field}")

    if user_input["operation"] not in ["query", "insert", "update"]:
        raise ValueError(f"Invalid operation: {user_input['operation']}")

    return True

Usage

Usage

try: validate_input(input_data["input"]) except ValueError as e: print(json.dumps({"error": str(e), "type": "ValidationError"})) sys.exit(1)
undefined
try: validate_input(input_data["input"]) except ValueError as e: print(json.dumps({"error": str(e), "type": "ValidationError"})) sys.exit(1)
undefined

Database Query Pattern

数据库查询模式

python
def safe_query(api_context, table_name, filters):
    """Execute a safe parameterized query"""
    # Build WHERE clause safely
    where_conditions = []
    for key, value in filters.items():
        # Simple validation
        if not key.isidentifier():
            raise ValueError(f"Invalid column name: {key}")
        where_conditions.append(f"{key} = '{value}'")

    where_clause = " AND ".join(where_conditions) if where_conditions else "1=1"
    sql = f"SELECT * FROM {table_name} WHERE {where_clause} LIMIT 100"

    return execute_sql(
        api_context["api_url"],
        api_context["api_token"],
        api_context["workspace_id"],
        api_context["stf_id"],
        sql
    )
python
def safe_query(api_context, table_name, filters):
    """Execute a safe parameterized query"""
    # Build WHERE clause safely
    where_conditions = []
    for key, value in filters.items():
        # Simple validation
        if not key.isidentifier():
            raise ValueError(f"Invalid column name: {key}")
        where_conditions.append(f"{key} = '{value}'")

    where_clause = " AND ".join(where_conditions) if where_conditions else "1=1"
    sql = f"SELECT * FROM {table_name} WHERE {where_clause} LIMIT 100"

    return execute_sql(
        api_context["api_url"],
        api_context["api_token"],
        api_context["workspace_id"],
        api_context["stf_id"],
        sql
    )

External API Pattern

外部API调用模式

python
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

def create_session():
    """Create session with retry logic"""
    session = requests.Session()
    retry = Retry(
        total=3,
        backoff_factor=0.3,
        status_forcelist=[500, 502, 503, 504]
    )
    adapter = HTTPAdapter(max_retries=retry)
    session.mount('http://', adapter)
    session.mount('https://', adapter)
    return session

def call_external_api(url, params):
    """Call external API with error handling"""
    session = create_session()
    try:
        response = session.get(url, params=params, timeout=10)
        response.raise_for_status()
        return response.json()
    except requests.Timeout:
        raise Exception("External API timeout")
    except requests.RequestException as e:
        raise Exception(f"External API error: {str(e)}")
python
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

def create_session():
    """Create session with retry logic"""
    session = requests.Session()
    retry = Retry(
        total=3,
        backoff_factor=0.3,
        status_forcelist=[500, 502, 503, 504]
    )
    adapter = HTTPAdapter(max_retries=retry)
    session.mount('http://', adapter)
    session.mount('https://', adapter)
    return session

def call_external_api(url, params):
    """Call external API with error handling"""
    session = create_session()
    try:
        response = session.get(url, params=params, timeout=10)
        response.raise_for_status()
        return response.json()
    except requests.Timeout:
        raise Exception("External API timeout")
    except requests.RequestException as e:
        raise Exception(f"External API error: {str(e)}")

Testing Locally

本地测试

Build and Test

构建与测试

bash
undefined
bash
undefined

Build image

Build image

docker build -t my-stf:latest .
docker build -t my-stf:latest .

Test with sample input

Test with sample input

echo '{ "workspace_id": "test-id", "stf_id": "test-stf-id", "caller": null, "api_url": "http://localhost:8080", "api_token": "test-token", "input": { "operation": "test" }, "sources": {} }' | docker run --rm -i my-stf:latest
undefined
echo '{ "workspace_id": "test-id", "stf_id": "test-stf-id", "caller": null, "api_url": "http://localhost:8080", "api_token": "test-token", "input": { "operation": "test" }, "sources": {} }' | docker run --rm -i my-stf:latest
undefined

Debug Mode

调试模式

bash
undefined
bash
undefined

Run with interactive shell

Run with interactive shell

docker run --rm -it --entrypoint /bin/bash my-stf:latest
docker run --rm -it --entrypoint /bin/bash my-stf:latest

Check image size

Check image size

docker images my-stf:latest
docker images my-stf:latest

Inspect logs

Inspect logs

docker run --rm -i my-stf:latest < input.json 2>&1 | tee output.log
undefined
docker run --rm -i my-stf:latest < input.json 2>&1 | tee output.log
undefined

Troubleshooting

故障排除

Issue: "Policy violation" error

问题："Policy violation" 错误

Cause: STF doesn't have permission to access the table.
Solution: Create policies:
javascript
// Create policy group
d6e_create_policy_group({ name: "my-stf-group" });

// Add STF to group
d6e_add_member_to_policy_group({
  policy_group_id: "{group_id}",
  member_type: "stf",
  member_id: "{stf_id}",
});

// Grant access
d6e_create_policy({
  policy_group_id: "{group_id}",
  table_name: "my_table",
  operation: "select",
  mode: "allow",
});
原因： STF没有访问该表的权限。
解决方案： 创建权限策略：
javascript
// Create policy group
d6e_create_policy_group({ name: "my-stf-group" });

// Add STF to group
d6e_add_member_to_policy_group({
  policy_group_id: "{group_id}",
  member_type: "stf",
  member_id: "{stf_id}",
});

// Grant access
d6e_create_policy({
  policy_group_id: "{group_id}",
  table_name: "my_table",
  operation: "select",
  mode: "allow",
});

Issue: Output not appearing in D6E

问题：输出未在D6E中显示

Cause: Output not in correct JSON format.
Solution: Always use 
{"output": {...}}
format:
python
undefined
原因： 输出格式不符合要求的JSON格式。
解决方案： 始终使用
{"output": {...}}
格式：
python
undefined

✅ Correct

✅ 正确格式

print(json.dumps({"output": {"status": "success"}}))
print(json.dumps({"output": {"status": "success"}}))

❌ Wrong

❌ 错误格式

print(json.dumps({"status": "success"}))
undefined
print(json.dumps({"status": "success"}))
undefined

Issue: "Image not found" in D6E

问题：D6E中提示"Image not found"

Cause: Image not accessible from D6E API server.
Solution:
  1. Publish to container registry (GitHub, Docker Hub)
  2. Or ensure same Docker daemon as D6E API server
原因： D6E API服务器无法访问该镜像。
解决方案：
  1. 将镜像发布到容器注册表（GitHub、Docker Hub等）
  2. 或确保与D6E API服务器使用同一个Docker守护进程

Issue: Large image size

问题：镜像体积过大

Solution: Use multi-stage builds:
dockerfile
undefined
解决方案： 使用多阶段构建：
dockerfile
undefined

Build stage

Build stage

FROM python:3.11 AS builder WORKDIR /app COPY requirements.txt . RUN pip install --user --no-cache-dir -r requirements.txt
FROM python:3.11 AS builder WORKDIR /app COPY requirements.txt . RUN pip install --user --no-cache-dir -r requirements.txt

Runtime stage

Runtime stage

FROM python:3.11-slim WORKDIR /app COPY --from=builder /root/.local /root/.local COPY main.py . ENV PATH=/root/.local/bin:$PATH ENTRYPOINT ["python3", "main.py"]
undefined
FROM python:3.11-slim WORKDIR /app COPY --from=builder /root/.local /root/.local COPY main.py . ENV PATH=/root/.local/bin:$PATH ENTRYPOINT ["python3", "main.py"]
undefined

File Requirements

文件要求

Every Docker STF should include:
my-stf/
├── main.py (or index.js, main.go)  # Entry point
├── Dockerfile                       # Container definition
├── requirements.txt (or package.json, go.mod)  # Dependencies
├── .dockerignore                    # Exclude files
└── README.md                        # Documentation
.dockerignore:
.git
.gitignore
*.md
tests/
__pycache__/
*.pyc
node_modules/
.env
每个Docker STF应包含以下文件：
my-stf/
├── main.py (或 index.js, main.go)  # 入口文件
├── Dockerfile                       # 容器定义文件
├── requirements.txt (或 package.json, go.mod)  # 依赖声明文件
├── .dockerignore                    # 排除文件配置
└── README.md                        # 文档说明
.dockerignore：
.git
.gitignore
*.md
tests/
__pycache__/
*.pyc
node_modules/
.env

README Template for D6E AI Agent Users

面向D6E AI Agent用户的README模板

When publishing a Docker STF, include a comprehensive README that enables D6E AI agents to automatically create and execute workflows. The README should follow this structure:
发布Docker STF时，请包含一份全面的README，使D6E AI Agent能够自动创建并执行工作流。README应遵循以下结构：

Required README Sections

必备README章节

  1. Title and Description - Clear name and purpose of the STF
  2. Docker Image URL - Full path (e.g., 
    ghcr.io/d6e-ai/stf-xxx:latest
    )
  3. LLM/AI Agent Usage Instructions - Step-by-step STF creation guide
  4. Supported Operations - Table of available operations with parameters
  5. Input/Output Examples - Concrete JSON examples for each operation
  6. AI Agent Prompts - Ready-to-use prompts for common tasks
  7. Troubleshooting - Common issues and solutions
  8. Local Build and Test - Commands for local development
  1. 标题与描述 - 清晰的STF名称和用途说明
  2. Docker镜像地址 - 完整路径（例如
    ghcr.io/d6e-ai/stf-xxx:latest
    ）
  3. LLM/AI Agent使用说明 - 分步的STF创建指南
  4. 支持的操作 - 可用操作的参数表格
  5. 输入/输出示例 - 每个操作的具体JSON示例
  6. AI Agent提示词 - 适用于常见任务的即用型提示词
  7. 故障排除 - 常见问题与解决方案
  8. 本地构建与测试 - 本地开发的命令说明

README Template

README模板

Use the following template for your Docker STF README:
markdown
undefined
使用以下模板创建Docker STF的README：
markdown
undefined

{STF Name}

{STF名称}

{Brief description of what this STF does}
Docker Image: 
ghcr.io/{org}/{stf-name}:latest
{该STF的简要功能描述}
Docker镜像：
ghcr.io/{组织名}/{stf名称}:latest

Usage for LLM/AI Agents

面向LLM/AI Agent的使用方法

To use this Docker image from a D6E AI agent, follow these steps to create and execute the STF.
要从D6E AI Agent使用此Docker镜像，请按照以下步骤创建并执行STF。

Step 1: Create the STF

步骤1：创建STF

javascript
d6e_create_stf({
  name: "{stf-name}",
  description: "{Description of the STF functionality}",
});
javascript
d6e_create_stf({
  name: "{stf名称}",
  description: "{STF功能描述}",
});

Step 2: Create the STF Version

步骤2：创建STF版本

javascript
d6e_create_stf_version({
  stf_id: "{stf_id from Step 1}",
  version: "1.0.0",
  runtime: "docker",
  code: '{"image":"ghcr.io/{org}/{stf-name}:latest"}',
});
Important: Always set 
runtime
to 
"docker"
and format the 
code
field as a JSON string: 
{"image":"ghcr.io/{org}/{stf-name}:latest"}
.
javascript
d6e_create_stf_version({
  stf_id: "{步骤1中的stf_id}",
  version: "1.0.0",
  runtime: "docker",
  code: '{"image":"ghcr.io/{组织名}/{stf名称}:latest"}',
});
重要提示：始终将
runtime
设置为
"docker"
，并将
code
字段格式化为JSON字符串：
'{"image":"ghcr.io/{组织名}/{stf名称}:latest"}'
。

Step 3: Create the Workflow

步骤3：创建工作流

javascript
d6e_create_workflow({
  name: "{stf-name}-workflow",
  input_steps: [],
  stf_steps: [
    {
      stf_id: "{stf_id}",
      version: "1.0.0",
    },
  ],
  effect_steps: [],
});
javascript
d6e_create_workflow({
  name: "{stf名称}-workflow",
  input_steps: [],
  stf_steps: [
    {
      stf_id: "{stf_id}",
      version: "1.0.0",
    },
  ],
  effect_steps: [],
});

Step 4: Execute the Workflow

步骤4：执行工作流

javascript
d6e_execute_workflow({
  workflow_id: "{workflow_id}",
  input: {
    operation: "{operation_name}",
    // ...operation-specific parameters
  },
});
javascript
d6e_execute_workflow({
  workflow_id: "{workflow_id}",
  input: {
    operation: "{操作名称}",
    // ...操作特定参数
  },
});

Supported Operations

支持的操作

OperationRequired ParametersOptionalDB RequiredDescription
{operation_1}
param1
, 
param2
optional1
❌/✅{Description}
{operation_2}
param1
-❌/✅{Description}
操作必填参数可选参数是否需要数据库描述
{操作1}
param1
, 
param2
optional1
❌/✅{描述}
{操作2}
param1
-❌/✅{描述}

Input/Output Examples

输入/输出示例

{Operation Name}

{操作名称}

Input:
json
{
  "operation": "{operation_name}",
  "param1": "value1",
  "param2": "value2"
}
Output:
json
{
  "output": {
    "status": "success",
    "operation": "{operation_name}",
    "data": {
      // ... result data
    }
  }
}
输入：
json
{
  "operation": "{操作名称}",
  "param1": "value1",
  "param2": "value2"
}
输出：
json
{
  "output": {
    "status": "success",
    "operation": "{操作名称}",
    "data": {
      // ... 结果数据
    }
  }
}

🤖 Prompts for AI Agents

🤖 面向AI Agent的提示词

Basic Prompt

基础提示词

Use the Docker skill for {task description} in D6E.

Docker Image: ghcr.io/{org}/{stf-name}:latest

Steps:
1. Create STF with d6e_create_stf (name: "{stf-name}")
2. Create STF version with d6e_create_stf_version:
   - runtime: "docker"
   - code: "{\"image\":\"ghcr.io/{org}/{stf-name}:latest\"}"
3. Create workflow with d6e_create_workflow
4. Execute with d6e_execute_workflow

Supported operations:
- "{operation_1}": {description} (required: {required_params})
- "{operation_2}": {description} (required: {required_params})

Start with {recommended_first_operation} to verify the setup.
在D6E中使用Docker技能完成{任务描述}。

Docker镜像：ghcr.io/{组织名}/{stf名称}:latest

步骤：
1. 使用d6e_create_stf创建STF（名称："{stf名称}"）
2. 使用d6e_create_stf_version创建STF版本：
   - runtime: "docker"
   - code: '{"image":"ghcr.io/{组织名}/{stf名称}:latest"}'
3. 使用d6e_create_workflow创建工作流
4. 使用d6e_execute_workflow执行工作流

支持的操作：
- "{操作1}"：{描述}（必填参数：{必填参数列表}）
- "{操作2}"：{描述}（必填参数：{必填参数列表}）

建议先执行{推荐的首个操作}以验证设置是否正确。

Task-Specific Prompt

任务特定提示词

{Specific task description}

Skill to use:
- Docker Image: ghcr.io/{org}/{stf-name}:latest
- Operation: {operation_name}

Parameters:
- param1: "value1"
- param2: "value2"

Include the following in the results:
- {Expected output item 1}
- {Expected output item 2}
{具体任务描述}

使用的技能：
- Docker镜像：ghcr.io/{组织名}/{stf名称}:latest
- 操作：{操作名称}

参数：
- param1: "value1"
- param2: "value2"

结果中需包含：
- {预期输出项1}
- {预期输出项2}

Complete Execution Prompt

完整执行提示词

{Complete workflow description}

Docker Image: ghcr.io/{org}/{stf-name}:latest

Execution steps:
1. Create STF (name: "{stf-name}", runtime: "docker")

2. {First operation description}:
   - operation: "{operation_1}"
   - param1: value1
   - param2: value2

3. {Second operation description}:
   - operation: "{operation_2}"
   - param1: value1

4. Display results:
   - {Output item 1}
   - {Output item 2}

{Additional instructions or requests}
{完整工作流描述}

Docker镜像：ghcr.io/{组织名}/{stf名称}:latest

执行步骤：
1. 创建STF（名称："{stf名称}", runtime: "docker"）

2. {第一个操作描述}：
   - operation: "{操作1}"
   - param1: value1
   - param2: value2

3. {第二个操作描述}：
   - operation: "{操作2}"
   - param1: value1

4. 显示结果：
   - {输出项1}
   - {输出项2}

{额外说明或要求}

Troubleshooting

故障排除

{Common Issue 1}

{常见问题1}

{Description and solution}
{问题描述与解决方案}

{Common Issue 2}

{常见问题2}

{Description and solution}
{问题描述与解决方案}

Local Build and Test

本地构建与测试

bash
undefined
bash
undefined

Build

构建镜像

docker build -t {stf-name}:latest .
docker build -t {stf名称}:latest .

Test

测试

echo '{ "workspace_id": "test-ws", "stf_id": "test-stf", "caller": null, "api_url": "http://localhost:8080", "api_token": "test-token", "input": { "operation": "{operation_name}", "param1": "value1" }, "sources": {} }' | docker run --rm -i {stf-name}:latest
undefined
echo '{ "workspace_id": "test-ws", "stf_id": "test-stf", "caller": null, "api_url": "http://localhost:8080", "api_token": "test-token", "input": { "operation": "{操作名称}", "param1": "value1" }, "sources": {} }' | docker run --rm -i {stf名称}:latest
undefined

Related Documentation

相关文档

  • Project README
  • {Additional documentation links}
undefined
  • 项目README
  • {其他文档链接}
undefined

Key Points for README Creation

README创建要点

  1. Explicit Docker Registration Instructions
    • Always specify 
      runtime: "docker"
    • Format 
      code
      as JSON string: 
      '{"image":"..."}'
    • Include the full image path with tag
  2. AI-Friendly Operation Tables
    • Use consistent table format
    • Clearly mark database requirements (❌/✅)
    • List all required and optional parameters
  3. Ready-to-Use Prompts
    • Provide multiple prompt examples (basic, specific, complete)
    • Include all necessary parameters in prompts
    • Suggest a recommended first operation for testing
  4. Clear Input/Output Examples
    • Show complete JSON structures
    • Include both success and error response examples
    • Document all possible output fields
  5. Self-Contained Instructions
    • Users should be able to copy the README and prompt to an AI agent
    • The AI agent should be able to execute without additional context
    • All steps should be clearly numbered and ordered
  1. 明确的Docker注册说明
    • 始终指定
      runtime: "docker"
    • 将
      code
      格式化为JSON字符串：
      '{"image":"..."}'
    • 包含完整的镜像路径和标签
  2. AI友好的操作表格
    • 使用统一的表格格式
    • 明确标记是否需要数据库（❌/✅）
    • 列出所有必填和可选参数
  3. 即用型提示词
    • 提供多种提示词示例（基础、特定任务、完整执行）
    • 在提示词中包含所有必要参数
    • 建议一个用于测试的推荐首个操作
  4. 清晰的输入/输出示例
    • 展示完整的JSON结构
    • 包含成功和错误响应示例
    • 记录所有可能的输出字段
  5. 自包含的说明
    • 用户应能够直接复制README和提示词给AI Agent
    • AI Agent无需额外上下文即可执行
    • 所有步骤应清晰编号并按顺序排列

Additional Resources

额外资源

For detailed information:
  • Complete API reference: reference.md
  • More implementation examples: examples.md
  • Quick start guide: ../docs/QUICKSTART.md
  • Testing guide: ../docs/TESTING.md
  • Publishing guide: ../docs/PUBLISHING.md
如需详细信息：
  • 完整API参考：reference.md
  • 更多实现示例：examples.md
  • 快速入门指南：../docs/QUICKSTART.md
  • 测试指南：../docs/TESTING.md
  • 发布指南：../docs/PUBLISHING.md ",