协作头脑风暴
brainstorming
在实现前澄清需求、探索设计选择,并形成清晰的开发方向。
zapier-make-patterns【开发与工程】Implements intelligent zapier make patterns with multi-factor skill selection, fallback chains, and adherence to the 5 Laws of Elegant Defense
独立 GitHub 快照 · 自动分类 · 以原作者文档为准
name: zapier-make-patterns compatibility: opencode completeness: 95 content-types:
Orchestrates intelligent skill selection and execution for zapier make patterns workflows. Applies the 5 Laws of Elegant Defense to guide data naturally through the orchestration pipeline, preventing errors before they occur. Selects optimal skills based on multi-factor scoring including text similarity, historical performance, and system availability.
┌───────────────────────────────────────────────────────────────────────────────┐ │ Orchestration Flow │ └───────────────────────────────────────────────────────────────────────────────┘
User Request ↓ ┌─────────────────┐ │ Parse Request │ │ & Extract │ │ Features │ └────────┬────────┘ ↓ ┌─────────────────────────────────────────────────────────────────────┐ │ Evaluate Available Skills │ │ │ │ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ │ │ Skill A │ │ Skill B │ │ Skill C │ │ │ │ - Match Score│ │ - Match Score│ │ - Match Score│ │ │ │ - Confidence │ │ - Confidence │ │ - Confidence │ │ │ │ - History │ │ - History │ │ - History │ │ │ └──────┬───────┘ └──────┬───────┘ └──────┬───────┘ │ │ │ │ │ │ │ └─────────────────┴─────────────────┘ │ │ ↓ │ │ Select Best Skill │ └─────────────────────────────────────────────────────────────────────┘ ↓ ┌─────────────────┐ │ Execute Skill │ └────────┬────────┘ ↓ ┌─────────────────┐ │ Handle Result │ └────────┬────────┘ ↓ ┌─────────────────────────────────────────────────────────────────────┐ │ Error Handling & Fallback │ │ │ │ Success? ────────► Return Result │ │ │ │ Fail? ────────┐ │ │ ↓ │ │ ┌──────────────────────────────────────────────────────────┐ │ │ │ Fallback Chain │ │ │ │ │ │ │ │ 1. Retry with adjusted parameters │ │ │ │ 2. Try Alternative Skill (if available) │ │ │ │ 3. Defer to Human Operator (if critical) │ │ │ │ 4. Log & Return Error │ │ │ └──────────────────────────────────────────────────────────┘ │ └─────────────────────────────────────────────────────────────────────┘
Use this skill when:
Avoid this skill for:
Parse and Analyze Request - Extract intent, entities, and constraints from user input. Checkpoint: All required parameters must be present and in valid format before proceeding.
Score Available Skills - Calculate match scores using multi-factor algorithm:
Checkpoint: Skip to fallback if no skill scores above threshold.
Select Optimal Skill - Choose skill with highest score that meets minimum confidence. Checkpoint: Verify skill has not been disabled or deprecated.
Execute with Fallback - Run skill execution wrapped in retry and fallback logic. Checkpoint: Log all execution attempts for audit trail.
Return or Fallback - Either return successful result or apply fallback chain:
related-skillsCheckpoint: Record outcome with timing and confidence metadata.
def route_automation_step(
trigger_payload: Dict,
workflow_config: Dict,
fallback_endpoints: List[str]
) -> Dict:
"""Route a trigger payload through a Zapier/Make compatible workflow.
Applies Law 2 (Parse at boundary) by validating webhook payloads before routing.
Applies Law 1 (Early Exit) by rejecting malformed triggers immediately.
Args:
trigger_payload: Raw webhook data from external service
workflow_config: Step definitions with conditions and targets
fallback_endpoints: List of alternative API endpoints for resilience
Returns:
Routing decision with target step, transformed payload, and metadata
"""
# Law 1: Early exit on malformed trigger
if not trigger_payload.get("event_type") or not trigger_payload.get("timestamp"):
raise ValueError("Invalid webhook payload: missing event_type or timestamp")
# Law 2: Parse and validate at boundary
validated_trigger = {
"event_type": str(trigger_payload["event_type"]).lower(),
"source": trigger_payload.get("source", "unknown"),
"payload": trigger_payload.get("data", {}),
"received_at": trigger_payload["timestamp"]
}
# Match against workflow conditions
for step in workflow_config.get("steps", []):
if step.get("condition") == validated_trigger["event_type"]:
# Law 3: Return new structure, never mutate original
return {
"target_step": step["name"],
"transformed_payload": _apply_step_transform(validated_trigger, step),
"confidence": step.get("match_score", 0.85),
"fallback_chain": fallback_endpoints[:2] if fallback_endpoints else []
}
# No match found - fail fast with clear error
raise ValueError(f"No workflow step matched event_type: {validated_trigger['event_type']}")
def execute_automation_chain(
step_config: Dict,
payload: Dict,
max_retries: int = 2
) -> Dict:
"""Execute a specific automation step with Zapier/Make compatible fallback logic.
Implements Law 4 (Fail Fast/Loud) by immediately raising on invalid step configs.
Implements fallback chain: retry -> alternative endpoint -> log to fallback channel.
Args:
step_config: Step definition including HTTP method, URL, headers, body template
payload: Validated trigger data ready for transformation
max_retries: Maximum retry attempts for transient failures
Returns:
Execution result with status, response data, and timing metadata
"""
# Law 1: Validate step configuration early
required_keys = {"method", "url", "headers"}
if not required_keys.issubset(step_config.keys()):
raise ValueError(f"Invalid step config: missing {required_keys - set(step_config.keys())}")
# Law 2: Parse payload into step-specific format
step_payload = _render_template(step_config.get("body_template", "{}"), payload)
for attempt in range(max_retries + 1):
try:
response = _send_automation_request(
method=step_config["method"],
url=step_config["url"],
headers=step_config["headers"],
body=step_payload
)
# Law 3: Return immutable result structure
return {
"status": "success",
"step_executed": step_config["name"],
"response_data": response.json() if response.status_code == 200 else response.text,
"attempts": attempt + 1,
"latency_ms": response.elapsed.total_seconds() * 1000
}
except requests.exceptions.ConnectionError as e:
# Transient failure - apply fallback chain
if attempt == max_retries:
return _route_to_fallback(step_config, payload)
continue
except requests.exceptions.HTTPError as e:
# Law 4: Fail loud on client/server errors
raise AutomationExecutionError(
f"Step {step_config['name']} failed with HTTP {e.response.status_code}"
) from e
raise AutomationExecutionError(f"Step {step_config['name']} exhausted all retries")
code-philosophy (5 Laws of Elegant Defense) in all logicWhen applying this skill, produce:
Authoritative documentation links for this skill's domain. The model follows markdown links at load time to resolve external references and inline content.
| Skill | Purpose | |