{
  "name": "Email Triage Agent",
  "nodes": [
    {
      "parameters": {
        "operation": "messageReceived",
        "pollTimes": {
          "interval": 5,
          "unit": "minutes"
        }
      },
      "name": "Gmail Trigger",
      "type": "n8n-nodes-base.gmailTrigger",
      "position": [250, 300]
    },
    {
      "parameters": {
        "model": "gpt-4",
        "prompt": "You are an email classifier. Analyze this email:\nSubject: {{$node[\"Gmail Trigger\"].json.subject}}\nBody: {{$node[\"Gmail Trigger\"].json.bodyPlain}}\nFrom: {{$node[\"Gmail Trigger\"].json.from}}\n\nReturn JSON only:\n{\n  \"intent\": \"complaint|inquiry|lead|other\",\n  \"sentiment\": \"negative|neutral|positive\",\n  \"urgency\": 1-5,\n  \"draft_reply\": \"brief draft reply\"\n}"
      },
      "name": "OpenAI Classify",
      "type": "n8n-nodes-base.openAi",
      "position": [450, 300]
    },
    {
      "parameters": {
        "conditions": {
          "string": [
            {
              "value1": "={{$json.urgency}}",
              "operation": "larger",
              "value2": 3
            }
          ]
        }
      },
      "name": "Urgency Check",
      "type": "n8n-nodes-base.if",
      "position": [650, 300]
    },
    {
      "parameters": {
        "operation": "send",
        "to": "={{$node[\"Gmail Trigger\"].json.from}}",
        "subject": "Re: {{$node[\"Gmail Trigger\"].json.subject}}",
        "body": "={{$json.draft_reply}}\\n\\n*This draft was generated by AI. Please review before sending.*"
      },
      "name": "Create Draft",
      "type": "n8n-nodes-base.gmail",
      "position": [850, 250]
    },
    {
      "parameters": {
        "operation": "create",
        "recordType": "lead",
        "fields": {
          "email": "={{$node[\"Gmail Trigger\"].json.from}}",
          "status": "={{$json.intent}}",
          "priority": "={{$json.urgency}}"
        }
      },
      "name": "Sync to CRM",
      "type": "n8n-nodes-base.salesforce",
      "position": [850, 350]
    }
  ]
}

#!/usr/bin/env python3
"""
Script: Research & Write Crew (CrewAI)
Search Intent: "CrewAI lead researcher and blog writer script"
Requirements: crewai, langchain, openai, firecrawl-python
Usage: python 02_research_write_crewai.py --topic "AI agents 2026" --output article.md
"""

import os
import argparse
from crewai import Agent, Task, Crew
from crewai.tools import tool
from langchain_openai import ChatOpenAI
from firecrawl import FirecrawlApp

# ============ CONFIGURATION ============
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
FIRECRAWL_API_KEY = os.getenv("FIRECRAWL_API_KEY")

if not OPENAI_API_KEY:
    raise ValueError("Set OPENAI_API_KEY environment variable")

# ============ CUSTOM TOOLS ============
@tool("Web Search")
def web_search(query: str) -> str:
    """Search the web for current information using Firecrawl."""
    app = FirecrawlApp(api_key=FIRECRAWL_API_KEY)
    result = app.search(query, limit=5)
    return "\n\n".join([r.get("markdown", "") for r in result])

# ============ AGENTS ============
researcher = Agent(
    role="Lead Researcher",
    goal="Find authoritative, recent information about {topic}",
    backstory="You are a senior research analyst. You verify facts, cite sources, and extract key insights.",
    tools=[web_search],
    llm=ChatOpenAI(model="gpt-4", temperature=0.3),
    allow_delegation=False,
    verbose=True
)

writer = Agent(
    role="Senior Content Writer",
    goal="Transform research into engaging, SEO-optimized blog posts",
    backstory="You write for tech professionals. Your tone is authoritative but conversational. You use headers, lists, and examples.",
    llm=ChatOpenAI(model="gpt-4", temperature=0.7),
    allow_delegation=False,
    verbose=True
)

# ============ TASKS ============
research_task = Task(
    description="""Research the following topic thoroughly: {topic}
    Find:
    1. Current trends and statistics (2026 data)
    2. Key tools and platforms professionals are using
    3. Common challenges and solutions
    4. Expert opinions or case studies
    
    Provide your findings in a structured format with citations.""",
    expected_output="Structured research brief with sources",
    agent=researcher
)

writing_task = Task(
    description="""Write a 2000-word blog post based on the research provided.
    
    Structure:
    - Compelling headline
    - Introduction (hook + what readers will learn)
    - 3-5 main sections with headers
    - Practical examples or code snippets
    - Conclusion + call to action
    
    Use the researcher's findings. Write for professionals, not beginners.""",
    expected_output="Complete blog post in Markdown",
    agent=writer,
    context=[research_task]
)

# ============ CREW ============
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task],
    verbose=2
)

# ============ MAIN ============
if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--topic", required=True, help="Topic to research and write about")
    parser.add_argument("--output", default="article.md", help="Output file path")
    args = parser.parse_args()
    
    result = crew.kickoff(inputs={"topic": args.topic})
    
    with open(args.output, "w") as f:
        f.write(result)
    
    print(f"✅ Article written to {args.output}")

#!/usr/bin/env python3
"""
Script: Stateful LangGraph Agent with Checkpointing
Search Intent: "LangGraph stateful agent with persistent checkpointing python"
Requirements: langgraph, langchain, redis, openai
Usage: python 03_stateful_langgraph.py --session-id user123
"""

import os
import argparse
from typing import TypedDict, Annotated, List
from langgraph.graph import StateGraph, END
from langgraph.checkpoint import MemorySaver
from langchain_openai import ChatOpenAI
from langchain.tools import tool
from langchain.agents import create_openai_tools_agent

# ============ STATE DEFINITION ============
class AgentState(TypedDict):
    messages: Annotated[List, "Conversation history"]
    task_status: str
    pending_actions: List[str]
    user_preferences: dict
    iteration_count: int

# ============ TOOLS ============
@tool
def update_memory(key: str, value: str) -> str:
    """Store user preferences or context for future sessions."""
    # In production, use Redis or PostgreSQL
    print(f"💾 Memory updated: {key} = {value}")
    return f"Stored {key}: {value}"

@tool
def execute_action(action: str) -> str:
    """Execute a scheduled action (send email, create task, etc.)."""
    print(f"⚡ Executing: {action}")
    return f"Action completed: {action}"

# ============ NODES ============
def call_model(state: AgentState):
    """Primary LLM node with access to conversation history."""
    llm = ChatOpenAI(model="gpt-4", temperature=0.7)
    # Simplified for brevity — full version includes prompt templates
    return {"messages": state["messages"] + ["Assistant response"]}

def check_backtrack(state: AgentState):
    """If task fails, return to previous step."""
    if state.get("task_status") == "failed":
        return "backtrack"
    return "continue"

def should_continue(state: AgentState):
    """Determine next step based on state."""
    if state.get("iteration_count", 0) > 5:
        return "end"
    return "call_model"

# ============ GRAPH BUILD ============
def build_agent_graph():
    workflow = StateGraph(AgentState)
    
    # Add nodes
    workflow.add_node("call_model", call_model)
    
    # Add conditional edges
    workflow.set_entry_point("call_model")
    workflow.add_conditional_edges(
        "call_model",
        should_continue,
        {
            "call_model": "call_model",
            "end": END
        }
    )
    
    # Add checkpointing for persistence
    memory = MemorySaver()
    app = workflow.compile(checkpointer=memory)
    
    return app

# ============ MAIN ============
if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--session-id", required=True, help="User session ID for persistence")
    args = parser.parse_args()
    
    app = build_agent_graph()
    
    config = {"configurable": {"thread_id": args.session_id}}
    
    print(f"✅ Agent ready. Session ID: {args.session_id}")
    print("Type your message (type 'exit' to quit):")
    
    while True:
        user_input = input("\n🧑 You: ")
        if user_input.lower() == "exit":
            break
        
        result = app.invoke(
            {"messages": [("user", user_input)], "iteration_count": 0},
            config=config
        )
        print(f"🤖 Agent: {result['messages'][-1].content}")

#!/usr/bin/env python3
"""
Script: Production Guardrail with PydanticAI
Search Intent: "PydanticAI typed agent response validation script"
Requirements: pydantic-ai, openai, tenacity
Usage: python 04_guardrail_pydantic.py --text "Customer email text"
"""

import os
import argparse
from typing import Literal
from pydantic import BaseModel, Field
from pydantic_ai import Agent
from tenacity import retry, stop_after_attempt, wait_exponential

# ============ TYPED RESPONSE SCHEMA ============
class EmailAnalysis(BaseModel):
    """Strictly typed response from the AI agent."""
    intent: Literal["complaint", "inquiry", "lead", "other"] = Field(
        description="Classification of email intent"
    )
    sentiment: Literal["negative", "neutral", "positive"] = Field(
        description="Emotional tone of the email"
    )
    urgency: int = Field(ge=1, le=5, description="Urgency score 1-5")
    summary: str = Field(max_length=200, description="One-sentence summary")
    requires_human: bool = Field(description="True if human review needed")
    draft_reply: str = Field(max_length=500, description="Suggested draft reply")

# ============ AGENT WITH GUARDRAILS ============
agent = Agent(
    model="openai:gpt-4",
    result_type=EmailAnalysis,
    system_prompt="""You are an email triage specialist.
    Analyze the email and return structured data.
    Always return valid JSON matching the schema.
    If uncertain, mark requires_human=true."""
)

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10),
    reraise=True
)
def analyze_with_retry(email_text: str) -> EmailAnalysis:
    """Analyze email with automatic retry on validation failure."""
    try:
        result = agent.run_sync(email_text)
        return result.data
    except Exception as e:
        print(f"⚠️ Validation failed: {e}. Retrying...")
        raise

# ============ MAIN ============
if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--text", required=True, help="Email text to analyze")
    parser.add_argument("--json", action="store_true", help="Output as JSON")
    args = parser.parse_args()
    
    try:
        analysis = analyze_with_retry(args.text)
        
        if args.json:
            print(analysis.model_dump_json(indent=2))
        else:
            print(f"\n📧 Analysis Results:")
            print(f"   Intent: {analysis.intent}")
            print(f"   Sentiment: {analysis.sentiment}")
            print(f"   Urgency: {analysis.urgency}/5")
            print(f"   Requires Human: {'✅' if analysis.requires_human else '❌'}")
            print(f"\n   Summary: {analysis.summary}")
            print(f"\n   Draft Reply:\n   {analysis.draft_reply}")
            
    except Exception as e:
        print(f"❌ Failed after retries: {e}")

#!/usr/bin/env python3
"""
Script: OpenClaw Local Controller
Search Intent: "OpenClaw terminal agent script for file management"
Requirements: openclaw, python-dotenv
Usage: python 05_openclaw_local.py --command "list all Python files in /projects"
"""

import os
import argparse
import subprocess
from pathlib import Path
from typing import List

# ============ SAFETY CONFIGURATION ============
ALLOWED_COMMANDS = [
    "ls", "cat", "find", "grep", "head", "tail",
    "mkdir", "cp", "mv", "rm", "pwd", "echo"
]

ALLOWED_PATHS = [
    os.path.expanduser("~/Documents"),
    os.path.expanduser("~/Projects"),
    os.path.expanduser("~/Downloads")
]

class OpenClawAgent:
    def __init__(self, dry_run: bool = True):
        self.dry_run = dry_run
        self.history = []
    
    def is_safe(self, command: str, path: str) -> bool:
        """Check if command and path are allowed."""
        cmd_base = command.split()[0]
        if cmd_base not in ALLOWED_COMMANDS:
            print(f"❌ Command '{cmd_base}' not allowed")
            return False
        
        for allowed_path in ALLOWED_PATHS:
            if path.startswith(allowed_path):
                return True
        
        print(f"❌ Path '{path}' not in allowed directories")
        return False
    
    def execute(self, command: str, path: str) -> str:
        """Execute a shell command with safety checks."""
        if not self.is_safe(command, path):
            return "Security check failed. Command not executed."
        
        self.history.append({"command": command, "path": path})
        
        if self.dry_run:
            return f"[DRY RUN] Would execute: {command} in {path}"
        
        try:
            result = subprocess.run(
                command,
                shell=True,
                cwd=path,
                capture_output=True,
                text=True,
                timeout=30
            )
            return result.stdout or result.stderr or "✅ Command completed."
        except Exception as e:
            return f"❌ Error: {e}"
    
    def list_files(self, path: str, pattern: str = "*") -> List[str]:
        """List files matching pattern in directory."""
        path_obj = Path(path).expanduser()
        if not path_obj.exists():
            return [f"Path not found: {path}"]
        
        files = list(path_obj.glob(pattern))
        return [str(f.relative_to(path_obj)) for f in files]

# ============ AI INTEGRATION ============
def get_ai_response(prompt: str) -> str:
    """Call LLM to interpret user request and generate command."""
    # In production, call OpenAI/Claude API
    # Simplified for example
    if "list" in prompt.lower() and "python" in prompt.lower():
        return "ls *.py"
    elif "find" in prompt.lower() and "log" in prompt.lower():
        return "find . -name '*.log' -type f"
    else:
        return "ls -la"

# ============ MAIN ============
if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--command", help="Direct shell command")
    parser.add_argument("--prompt", help="Natural language prompt")
    parser.add_argument("--path", default="~/Projects", help="Working directory")
    parser.add_argument("--dry-run", action="store_true", default=True)
    parser.add_argument("--execute", action="store_true", help="Actually execute (disable dry-run)")
    args = parser.parse_args()
    
    agent = OpenClawAgent(dry_run=not args.execute)
    
    if args.command:
        result = agent.execute(args.command, os.path.expanduser(args.path))
        print(result)
    elif args.prompt:
        command = get_ai_response(args.prompt)
        print(f"🤖 AI generated: {command}")
        result = agent.execute(command, os.path.expanduser(args.path))
        print(result)
    else:
        print("OpenClaw Local Controller")
        print("Usage:")
        print("  --command 'ls -la'        Execute shell command")
        print("  --prompt 'list python files'   Natural language")
        print("  --execute                  Actually run (default is dry-run)")
        print("  --path ~/Projects          Working directory")