Activepieces：开源自动化工具

export const activepiecesPiece = createPiece({  name: 'activepieces',  displayName: 'Activepieces',  logoUrl: 'https://cdn.activepieces.com/piece.png',  authors: ['Community'],  triggers: [createTrigger({...})],  actions: [createAction({...})]});

import requests def call_llm_via_activepieces(api_url, api_key, prompt, context_data):    """通过Activepieces AI Piece调用LLM"""    rendered_prompt = prompt.format(**context_data)    response = requests.post(        api_url,        headers={"Authorization": f"Bearer {api_key}"},        json={"model": "gpt-4", "messages": [{"role": "user", "content": rendered_prompt}]}    )    return response.json()["choices"][0]["message"]["content"]

version: '3.8'services:  activepieces:    image: activepieces/activepieces:latest    ports:      - "8080:80"    environment:      - AP_DB_TYPE=POSTGRES      - AP_POSTGRES_HOST=postgres      - AP_REDIS_HOST=redis      - AP_ENCRYPTION_KEY=${ENCRYPTION_KEY}    depends_on:      - postgres      - redis

import requestsimport jsonfrom datetime import datetime, timedelta AP_BASE_URL = "http://localhost:8080/api/v1"AP_API_KEY = "your-api-key-here" class ActivepiecesFlowRunner:    def __init__(self, base_url, api_key):        self.base_url = base_url        self.headers = {            "Authorization": f"Bearer {api_key}",            "Content-Type": "application/json"        }        def list_flows(self):        resp = requests.get(            f"{self.base_url}/flows",            headers=self.headers        )        resp.raise_for_status()        return resp.json()        def get_flow(self, flow_id):        resp = requests.get(            f"{self.base_url}/flows/{flow_id}",            headers=self.headers        )        resp.raise_for_status()        return resp.json()        def execute_flow(self, flow_id, payload=None):        data = {"payload": payload or {}}        resp = requests.post(            f"{self.base_url}/flows/{flow_id}/execute",            headers=self.headers,            json=data        )        resp.raise_for_status()        return resp.json()        def get_run_history(self, flow_id, limit=10):        resp = requests.get(            f"{self.base_url}/flows/{flow_id}/runs",            headers=self.headers,            params={"limit": limit}        )        resp.raise_for_status()        return resp.json() class AIAutomationBuilder:    def __init__(self, runner):        self.runner = runner        def create_email_summary_flow(self, flow_name, openai_api_key):        flow_config = {            "displayName": flow_name,            "trigger": {                "type": "PIECE_TRIGGER",                "settings": {                    "pieceName": "gmail",                    "pieceVersion": "0.1.0",                    "triggerName": "new_email",                    "input": {"label": "inbox"}                }            },            "steps": [                {                    "name": "ai_summarize",                    "type": "PIECE",                    "settings": {                        "pieceName": "openai",                        "pieceVersion": "0.2.0",                        "actionName": "send_prompt",                        "input": {                            "model": "gpt-4",                            "prompt": "Summarize: {{trigger.subject}} - {{trigger.body}}",                            "apiKey": openai_api_key                        }                    }                },                {                    "name": "slack_notify",                    "type": "PIECE",                    "settings": {                        "pieceName": "slack",                        "pieceVersion": "0.1.0",                        "actionName": "send_message",                        "input": {                            "channel": "#ai-summaries",                            "text": "New Email: {{ai_summarize.response}}"                        }                    }                }            ]        }        return flow_config        def simulate_ai_workflow(self, email_subject, email_body):        context = {            "trigger": {                "subject": email_subject,                "body": email_body,                "from": "sender@example.com",                "received_at": datetime.now().isoformat()            }        }        ai_prompt = f"Summarize: {email_body[:200]}"        ai_response = f"Key points:\n1. {email_subject}\n2. Action required\n3. Deadline: {email_body[:50]}"        context["ai_summarize"] = {"response": ai_response}        slack_message = f"Email Summary:\n{ai_response}"        context["slack_notify"] = {"sent": True, "message": slack_message}        return context if __name__ == "__main__":    runner = ActivepiecesFlowRunner(AP_BASE_URL, AP_API_KEY)    print("=== Published Flows ===")    flows = runner.list_flows()    for flow in flows.get("data", []):        print(f"  - {flow['displayName']} (ID: {flow['id']})")    builder = AIAutomationBuilder(runner)    print("\n=== Simulating AI Workflow ===")    result = builder.simulate_ai_workflow(        email_subject="Q3 Product Roadmap Review",        email_body="Hi team, please review the Q3 roadmap and provide feedback by Friday."    )    for step_name, step_data in result.items():        if step_name != "trigger":            print(f"\n[{step_name}] Done:")            print(json.dumps(step_data, indent=2, ensure_ascii=False))    print("\n=== AI Flow Config ===")    flow_config = builder.create_email_summary_flow(        "Email AI Summarizer",        "sk-your-openai-key"    )    print(json.dumps(flow_config, indent=2, ensure_ascii=False))