AI Agent Actions

Build AI agents that can take real-world actions by integrating with external services like Slack, JIRA, GitHub, and deployment systems through Hyrex task orchestration.

Overview

This example demonstrates how to create actionable AI agents:

Multi-service integration - Slack notifications, JIRA tickets, deployments
Action routing - Dispatch different actions based on agent requests
Bulk operations - Process multiple agent actions efficiently
Security & validation - Ensure agents can only perform authorized actions
Audit logging - Track all agent-initiated actions

Perfect for building AI assistants, automated workflows, or agent-driven DevOps systems.

Task Definitions

from hyrex import HyrexRegistry
import requests
import json
import os

hy = HyrexRegistry()

@hy.task
def send_slack_notification(message: str, channel: str):
    """Send notification to Slack channel"""
    payload = {
        "text": message,
        "channel": channel
    }
    
    response = requests.post(
        os.environ.get("SLACK_WEBHOOK_URL"),
        json=payload
    )
    
    if response.status_code != 200:
        raise Exception(f"Slack notification failed: {response.text}")
    
    return {"status": "sent", "channel": channel, "message_length": len(message)}

@hy.task
def create_jira_ticket(title: str, description: str, assignee: str, priority: str = "Medium"):
    """Create a new JIRA ticket"""
    ticket_data = {
        "fields": {
            "project": {"key": os.environ.get("JIRA_PROJECT_KEY", "PROJ")},
            "summary": title,
            "description": {
                "content": [
                    {
                        "content": [{"text": description, "type": "text"}],
                        "type": "paragraph"
                    }
                ],
                "type": "doc",
                "version": 1
            },
            "issuetype": {"name": "Task"},
            "assignee": {"displayName": assignee},
            "priority": {"name": priority}
        }
    }
    
    response = requests.post(
        f"{os.environ.get('JIRA_BASE_URL')}/rest/api/3/issue",
        json=ticket_data,
        auth=(os.environ.get("JIRA_EMAIL"), os.environ.get("JIRA_TOKEN"))
    )
    
    if response.status_code not in [200, 201]:
        raise Exception(f"JIRA ticket creation failed: {response.text}")
    
    ticket = response.json()
    return {
        "ticket_id": ticket["key"],
        "ticket_url": f"{os.environ.get('JIRA_BASE_URL')}/browse/{ticket['key']}",
        "title": title,
        "assignee": assignee
    }

@hy.task
def create_github_issue(repo: str, title: str, body: str, labels: list = None):
    """Create GitHub issue"""
    issue_data = {
        "title": title,
        "body": body,
        "labels": labels or []
    }
    
    response = requests.post(
        f"https://api.github.com/repos/{repo}/issues",
        json=issue_data,
        headers={
            "Authorization": f"token {os.environ.get('GITHUB_TOKEN')}",
            "Accept": "application/vnd.github.v3+json"
        }
    )
    
    if response.status_code != 201:
        raise Exception(f"GitHub issue creation failed: {response.text}")
    
    issue = response.json()
    return {
        "issue_id": issue["number"],
        "issue_url": issue["html_url"],
        "title": title,
        "repo": repo
    }

@hy.task
def deploy_service(service_name: str, environment: str = "production", version: str = "latest"):
    """Deploy service to specified environment"""
    # Integration with your deployment system (K8s, Docker, etc.)
    deployment_config = {
        "service": service_name,
        "environment": environment,
        "version": version,
        "triggered_by": "ai_agent"
    }
    
    # Example: Trigger deployment pipeline
    response = requests.post(
        f"{os.environ.get('DEPLOYMENT_API_URL')}/deploy",
        json=deployment_config,
        headers={"Authorization": f"Bearer {os.environ.get('DEPLOYMENT_TOKEN')}"}
    )
    
    if response.status_code != 200:
        raise Exception(f"Deployment failed: {response.text}")
    
    deployment = response.json()
    return {
        "deployment_id": deployment.get("id"),
        "service": service_name,
        "environment": environment,
        "status": "initiated"
    }

@hy.task
def send_email_alert(recipients: list, subject: str, body: str, priority: str = "normal"):
    """Send email alert to specified recipients"""
    email_data = {
        "recipients": recipients,
        "subject": subject,
        "body": body,
        "priority": priority
    }
    
    # Integration with your email service
    response = requests.post(
        f"{os.environ.get('EMAIL_SERVICE_URL')}/send",
        json=email_data,
        headers={"Authorization": f"Bearer {os.environ.get('EMAIL_SERVICE_TOKEN')}"}
    )
    
    if response.status_code != 200:
        raise Exception(f"Email sending failed: {response.text}")
    
    return {
        "recipients": recipients,
        "subject": subject,
        "status": "sent"
    }

@hy.task
def log_agent_action(agent_id: str, action_type: str, payload: dict, result: dict = None):
    """Log agent action for audit purposes"""
    log_entry = {
        "timestamp": datetime.now().isoformat(),
        "agent_id": agent_id,
        "action_type": action_type,
        "payload": payload,
        "result": result,
        "status": "success" if result else "attempted"
    }
    
    # Store in your logging system
    save_to_audit_log(log_entry)
    return log_entry

@hy.task
def process_ai_agent_action(action_type: str, payload: dict, agent_id: str = "unknown"):
    """Process action requested by AI agent"""
    result = None
    
    try:
        if action_type == "notify_slack":
            result = send_slack_notification.send(
                payload["message"], 
                payload["channel"]
            ).get()
            
        elif action_type == "create_jira_ticket":
            result = create_jira_ticket.send(
                payload["title"],
                payload["description"],
                payload["assignee"],
                payload.get("priority", "Medium")
            ).get()
            
        elif action_type == "create_github_issue":
            result = create_github_issue.send(
                payload["repo"],
                payload["title"],
                payload["body"],
                payload.get("labels", [])
            ).get()
            
        elif action_type == "deploy_service":
            result = deploy_service.send(
                payload["service_name"],
                payload.get("environment", "production"),
                payload.get("version", "latest")
            ).get()
            
        elif action_type == "send_email":
            result = send_email_alert.send(
                payload["recipients"],
                payload["subject"],
                payload["body"],
                payload.get("priority", "normal")
            ).get()
            
        else:
            raise ValueError(f"Unsupported action type: {action_type}")
        
        # Log successful action
        log_agent_action.send(agent_id, action_type, payload, result)
        
        return {
            "success": True,
            "action_type": action_type,
            "result": result,
            "agent_id": agent_id
        }
        
    except Exception as e:
        # Log failed action
        log_agent_action.send(agent_id, action_type, payload, {"error": str(e)})
        raise e

import { HyrexRegistry } from 'hyrex';
import axios from 'axios';

const hy = new HyrexRegistry();

const sendSlackNotification = hy.task({
    name: "sendSlackNotification",
    func: async (message: string, channel: string) => {
        const payload = {
            text: message,
            channel: channel
        };
        
        const response = await axios.post(
            process.env.SLACK_WEBHOOK_URL!,
            payload
        );
        
        return { 
            status: "sent", 
            channel, 
            messageLength: message.length 
        };
    }
});

const createJiraTicket = hy.task({
    name: "createJiraTicket", 
    func: async (title: string, description: string, assignee: string, priority: string = "Medium") => {
        const ticketData = {
            fields: {
                project: { key: process.env.JIRA_PROJECT_KEY || "PROJ" },
                summary: title,
                description: {
                    content: [
                        {
                            content: [{ text: description, type: "text" }],
                            type: "paragraph"
                        }
                    ],
                    type: "doc",
                    version: 1
                },
                issuetype: { name: "Task" },
                assignee: { displayName: assignee },
                priority: { name: priority }
            }
        };
        
        const response = await axios.post(
            `${process.env.JIRA_BASE_URL}/rest/api/3/issue`,
            ticketData,
            {
                auth: {
                    username: process.env.JIRA_EMAIL!,
                    password: process.env.JIRA_TOKEN!
                }
            }
        );
        
        const ticket = response.data;
        return {
            ticketId: ticket.key,
            ticketUrl: `${process.env.JIRA_BASE_URL}/browse/${ticket.key}`,
            title,
            assignee
        };
    }
});

const createGithubIssue = hy.task({
    name: "createGithubIssue",
    func: async (repo: string, title: string, body: string, labels: string[] = []) => {
        const issueData = {
            title,
            body,
            labels
        };
        
        const response = await axios.post(
            `https://api.github.com/repos/${repo}/issues`,
            issueData,
            {
                headers: {
                    "Authorization": `token ${process.env.GITHUB_TOKEN}`,
                    "Accept": "application/vnd.github.v3+json"
                }
            }
        );
        
        const issue = response.data;
        return {
            issueId: issue.number,
            issueUrl: issue.html_url,
            title,
            repo
        };
    }
});

const deployService = hy.task({
    name: "deployService",
    func: async (serviceName: string, environment: string = "production", version: string = "latest") => {
        const deploymentConfig = {
            service: serviceName,
            environment,
            version,
            triggeredBy: "ai_agent"
        };
        
        const response = await axios.post(
            `${process.env.DEPLOYMENT_API_URL}/deploy`,
            deploymentConfig,
            {
                headers: {
                    "Authorization": `Bearer ${process.env.DEPLOYMENT_TOKEN}`
                }
            }
        );
        
        const deployment = response.data;
        return {
            deploymentId: deployment.id,
            service: serviceName,
            environment,
            status: "initiated"
        };
    }
});

const sendEmailAlert = hy.task({
    name: "sendEmailAlert",
    func: async (recipients: string[], subject: string, body: string, priority: string = "normal") => {
        const emailData = {
            recipients,
            subject,
            body,
            priority
        };
        
        const response = await axios.post(
            `${process.env.EMAIL_SERVICE_URL}/send`,
            emailData,
            {
                headers: {
                    "Authorization": `Bearer ${process.env.EMAIL_SERVICE_TOKEN}`
                }
            }
        );
        
        return {
            recipients,
            subject,
            status: "sent"
        };
    }
});

const logAgentAction = hy.task({
    name: "logAgentAction",
    func: async (agentId: string, actionType: string, payload: any, result: any = null) => {
        const logEntry = {
            timestamp: new Date().toISOString(),
            agentId,
            actionType,
            payload,
            result,
            status: result ? "success" : "attempted"
        };
        
        // Store in your logging system
        await saveToAuditLog(logEntry);
        return logEntry;
    }
});

const processAIAgentAction = hy.task({
    name: "processAIAgentAction",
    func: async (actionType: string, payload: any, agentId: string = "unknown") => {
        let result: any = null;
        
        try {
            switch (actionType) {
                case "notify_slack":
                    const slackTask = await sendSlackNotification.send(payload.message, payload.channel);
                    result = await slackTask.get();
                    break;
                    
                case "create_jira_ticket":
                    const jiraTask = await createJiraTicket.send(
                        payload.title,
                        payload.description,
                        payload.assignee,
                        payload.priority || "Medium"
                    );
                    result = await jiraTask.get();
                    break;
                    
                case "create_github_issue":
                    const githubTask = await createGithubIssue.send(
                        payload.repo,
                        payload.title,
                        payload.body,
                        payload.labels || []
                    );
                    result = await githubTask.get();
                    break;
                    
                case "deploy_service":
                    const deployTask = await deployService.send(
                        payload.serviceName,
                        payload.environment || "production",
                        payload.version || "latest"
                    );
                    result = await deployTask.get();
                    break;
                    
                case "send_email":
                    const emailTask = await sendEmailAlert.send(
                        payload.recipients,
                        payload.subject,
                        payload.body,
                        payload.priority || "normal"
                    );
                    result = await emailTask.get();
                    break;
                    
                default:
                    throw new Error(`Unsupported action type: ${actionType}`);
            }
            
            // Log successful action
            await logAgentAction.send(agentId, actionType, payload, result);
            
            return {
                success: true,
                actionType,
                result,
                agentId
            };
            
        } catch (error: any) {
            // Log failed action
            await logAgentAction.send(agentId, actionType, payload, { error: error.message });
            throw error;
        }
    }
});

REST API Endpoints

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional
from .tasks import process_ai_agent_action

app = FastAPI()

class AgentActionRequest(BaseModel):
    action_type: str
    payload: dict
    agent_id: Optional[str] = None
    
class BulkActionRequest(BaseModel):
    actions: List[AgentActionRequest]

@app.post("/agent/action")
async def execute_agent_action(request: AgentActionRequest):
    """Execute a single agent action"""
    try:
        task = process_ai_agent_action.send(
            request.action_type, 
            request.payload,
            request.agent_id or "unknown"
        )
        
        return {
            "message": "Agent action queued successfully", 
            "task_id": task.task_id,
            "action_type": request.action_type,
            "agent_id": request.agent_id
        }
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))

@app.post("/agent/bulk-actions") 
async def execute_bulk_actions(request: BulkActionRequest):
    """Execute multiple agent actions"""
    task_ids = []
    
    for action in request.actions:
        try:
            task = process_ai_agent_action.send(
                action.action_type,
                action.payload,
                action.agent_id or "unknown"
            )
            task_ids.append({
                "task_id": task.task_id,
                "action_type": action.action_type,
                "agent_id": action.agent_id
            })
        except Exception as e:
            task_ids.append({
                "error": str(e),
                "action_type": action.action_type,
                "agent_id": action.agent_id
            })
    
    return {
        "message": f"Queued {len([t for t in task_ids if 'task_id' in t])} actions",
        "results": task_ids
    }

@app.get("/agent/actions/{task_id}/status")
async def get_action_status(task_id: str):
    """Get status of an agent action"""
    task = hy.get_task(task_id)
    
    return {
        "task_id": task_id,
        "status": task.status,
        "result": task.result if task.is_complete else None,
        "error": task.error if task.has_failed else None,
        "created_at": task.created_at,
        "completed_at": task.completed_at
    }

@app.get("/agent/{agent_id}/actions")
async def get_agent_actions(agent_id: str):
    """Get all actions for a specific agent"""
    # Retrieve from audit log
    actions = get_agent_actions_from_log(agent_id)
    return {
        "agent_id": agent_id,
        "actions": actions,
        "total_actions": len(actions)
    }

import express from 'express';
import { processAIAgentAction } from './tasks';

const app = express();
app.use(express.json());

interface AgentActionRequest {
    actionType: string;
    payload: any;
    agentId?: string;
}

interface BulkActionRequest {
    actions: AgentActionRequest[];
}

app.post('/agent/action', async (req, res) => {
    const { actionType, payload, agentId }: AgentActionRequest = req.body;
    
    try {
        const task = await processAIAgentAction.send(
            actionType,
            payload,
            agentId || "unknown"
        );
        
        res.json({
            message: 'Agent action queued successfully',
            taskId: task.taskId,
            actionType,
            agentId
        });
    } catch (error: any) {
        res.status(400).json({ error: error.message });
    }
});

app.post('/agent/bulk-actions', async (req, res) => {
    const { actions }: BulkActionRequest = req.body;
    
    const results = [];
    
    for (const action of actions) {
        try {
            const task = await processAIAgentAction.send(
                action.actionType,
                action.payload,
                action.agentId || "unknown"
            );
            results.push({
                taskId: task.taskId,
                actionType: action.actionType,
                agentId: action.agentId
            });
        } catch (error: any) {
            results.push({
                error: error.message,
                actionType: action.actionType,
                agentId: action.agentId
            });
        }
    }
    
    const successCount = results.filter(r => r.taskId).length;
    
    res.json({
        message: `Queued ${successCount} actions`,
        results
    });
});

app.get('/agent/actions/:taskId/status', async (req, res) => {
    const { taskId } = req.params;
    
    try {
        const task = await hy.getTask(taskId);
        
        res.json({
            taskId,
            status: task.status,
            result: task.isComplete ? task.result : null,
            error: task.hasFailed ? task.error : null,
            createdAt: task.createdAt,
            completedAt: task.completedAt
        });
    } catch (error: any) {
        res.status(404).json({ error: 'Task not found' });
    }
});

app.get('/agent/:agentId/actions', async (req, res) => {
    const { agentId } = req.params;
    
    try {
        // Retrieve from audit log
        const actions = await getAgentActionsFromLog(agentId);
        
        res.json({
            agentId,
            actions,
            totalActions: actions.length
        });
    } catch (error: any) {
        res.status(500).json({ error: error.message });
    }
});

Usage Examples

Single Agent Action

# Send Slack notification
curl -X POST http://localhost:8000/agent/action \
  -H "Content-Type: application/json" \
  -d '{
    "action_type": "notify_slack",
    "payload": {
      "message": "Deployment completed successfully!",
      "channel": "#deployments"
    },
    "agent_id": "deploy_bot_v1"
  }'

# Create JIRA ticket
curl -X POST http://localhost:8000/agent/action \
  -H "Content-Type: application/json" \
  -d '{
    "action_type": "create_jira_ticket",
    "payload": {
      "title": "Investigate high memory usage",
      "description": "Memory usage spiked to 95% at 2:30 PM",
      "assignee": "john.doe",
      "priority": "High"
    },
    "agent_id": "monitoring_agent"
  }'

# Deploy service
curl -X POST http://localhost:8000/agent/action \
  -H "Content-Type: application/json" \
  -d '{
    "action_type": "deploy_service",
    "payload": {
      "service_name": "user-service",
      "environment": "production",
      "version": "v1.2.3"
    },
    "agent_id": "deploy_agent"
  }'

Bulk Actions

curl -X POST http://localhost:8000/agent/bulk-actions \
  -H "Content-Type: application/json" \
  -d '{
    "actions": [
      {
        "action_type": "notify_slack",
        "payload": {
          "message": "Starting deployment sequence",
          "channel": "#deployments"
        }
      },
      {
        "action_type": "deploy_service",
        "payload": {
          "service_name": "api-gateway",
          "environment": "production"
        }
      },
      {
        "action_type": "send_email",
        "payload": {
          "recipients": ["team@company.com"],
          "subject": "Production Deployment Started",
          "body": "The production deployment has been initiated by the AI agent."
        }
      }
    ]
  }'

Advanced Agent Patterns

Conditional Actions

@hy.task
def process_conditional_action(condition: dict, actions: list):
    """Execute actions based on conditions"""
    if evaluate_condition(condition):
        for action in actions:
            process_ai_agent_action.send(
                action['type'], 
                action['payload'], 
                action.get('agent_id')
            )
        return {"executed": len(actions), "condition_met": True}
    else:
        return {"executed": 0, "condition_met": False}

Action Chains

@hy.task
def execute_action_chain(actions: list, agent_id: str):
    """Execute actions in sequence with dependency handling"""
    results = []
    
    for action in actions:
        try:
            result = process_ai_agent_action.send(
                action['type'], 
                action['payload'], 
                agent_id
            ).get()
            
            results.append(result)
            
            # Check if we should continue based on result
            if action.get('stop_on_failure') and not result['success']:
                break
                
        except Exception as e:
            results.append({"error": str(e), "action": action['type']})
            if action.get('stop_on_failure', True):
                break
    
    return {
        "chain_results": results,
        "completed_actions": len(results),
        "total_actions": len(actions)
    }

Permission-Based Actions

@hy.task
def validate_agent_permissions(agent_id: str, action_type: str):
    """Validate if agent has permission for action"""
    agent_permissions = get_agent_permissions(agent_id)
    
    if action_type not in agent_permissions:
        raise PermissionError(f"Agent {agent_id} not authorized for {action_type}")
    
    return True

@hy.task
def secure_process_action(action_type: str, payload: dict, agent_id: str):
    """Process action with permission validation"""
    # Validate permissions first
    validate_agent_permissions.send(agent_id, action_type).get()
    
    # Execute the action
    return process_ai_agent_action.send(action_type, payload, agent_id).get()

Security & Compliance

Environment Variables

# Slack Integration
SLACK_WEBHOOK_URL=https://hooks.slack.com/services/YOUR/WEBHOOK/URL

# JIRA Integration  
JIRA_BASE_URL=https://your-domain.atlassian.net
JIRA_EMAIL=your-email@company.com
JIRA_TOKEN=your-api-token
JIRA_PROJECT_KEY=PROJ

# GitHub Integration
GITHUB_TOKEN=ghp_your-token

# Deployment Integration
DEPLOYMENT_API_URL=https://your-deployment-api.com
DEPLOYMENT_TOKEN=your-deployment-token

# Email Service
EMAIL_SERVICE_URL=https://your-email-service.com
EMAIL_SERVICE_TOKEN=your-email-token

Action Validation

ACTION_SCHEMAS = {
    "notify_slack": {
        "required": ["message", "channel"],
        "properties": {
            "message": {"type": "string", "maxLength": 4000},
            "channel": {"type": "string", "pattern": "^#[a-z0-9_-]+$"}
        }
    },
    "create_jira_ticket": {
        "required": ["title", "description", "assignee"],
        "properties": {
            "title": {"type": "string", "maxLength": 255},
            "description": {"type": "string", "maxLength": 32767},
            "assignee": {"type": "string"},
            "priority": {"enum": ["Lowest", "Low", "Medium", "High", "Highest"]}
        }
    }
}

Production Considerations

Rate limiting: Prevent agents from overwhelming external services
Permission management: Implement role-based access control for agents
Audit logging: Track all agent actions for compliance and debugging
Error handling: Graceful failure handling with proper notifications
Service limits: Respect external API rate limits and quotas
Security: Secure storage of API keys and authentication tokens

Overview

SDK Reference

Examples

Production Guide

AI Agent Actions

Overview

Task Definitions

REST API Endpoints

Usage Examples

Single Agent Action

Bulk Actions

Advanced Agent Patterns

Conditional Actions

Action Chains

Permission-Based Actions

Security & Compliance

Environment Variables

Action Validation

Production Considerations

Next Steps

Error Handling

Context Engineering

​Overview

​Task Definitions

​REST API Endpoints

​Usage Examples

​Single Agent Action

​Bulk Actions

​Advanced Agent Patterns

​Conditional Actions

​Action Chains

​Permission-Based Actions

​Security & Compliance

​Environment Variables

​Action Validation

​Production Considerations

​Next Steps

Error Handling

Context Engineering

Overview

Task Definitions

REST API Endpoints

Usage Examples

Single Agent Action

Bulk Actions

Advanced Agent Patterns

Conditional Actions

Action Chains

Permission-Based Actions

Security & Compliance

Environment Variables

Action Validation

Production Considerations

Next Steps