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

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

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

docker build -t my-stf:latest .

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

undefined

Debug Mode

调试模式

bash

undefined

bash

undefined

Run with interactive shell

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

Check image size

docker images my-stf:latest

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"}}))

❌ 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:

Publish to container registry (GitHub, Docker Hub)
Or ensure same Docker daemon as D6E API server

原因： D6E API服务器无法访问该镜像。

解决方案：

将镜像发布到容器注册表（GitHub、Docker Hub等）
或确保与D6E API服务器使用同一个Docker守护进程

Issue: Large image size

问题：镜像体积过大

Solution: Use multi-stage builds:

dockerfile

undefined

解决方案： 使用多阶段构建：

dockerfile

undefined

Build stage

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

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章节

Title and Description - Clear name and purpose of the STF
Docker Image URL - Full path (e.g.,
```
ghcr.io/d6e-ai/stf-xxx:latest
```
)
LLM/AI Agent Usage Instructions - Step-by-step STF creation guide
Supported Operations - Table of available operations with parameters
Input/Output Examples - Concrete JSON examples for each operation
AI Agent Prompts - Ready-to-use prompts for common tasks
Troubleshooting - Common issues and solutions
Local Build and Test - Commands for local development

标题与描述 - 清晰的STF名称和用途说明
Docker镜像地址 - 完整路径（例如
```
ghcr.io/d6e-ai/stf-xxx:latest
```
）
LLM/AI Agent使用说明 - 分步的STF创建指南
支持的操作 - 可用操作的参数表格
输入/输出示例 - 每个操作的具体JSON示例
AI Agent提示词 - 适用于常见任务的即用型提示词
故障排除 - 常见问题与解决方案
本地构建与测试 - 本地开发的命令说明

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

"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

支持的操作

Operation	Required Parameters	Optional	DB Required	Description
`{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

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 ",