mirror of
https://github.com/MervinPraison/PraisonAI-Tools.git
synced 2026-11-07 05:56:17 -10:00
feat: Add n8n workflow automation integration (#19)
* feat: add n8n workflow automation integration (fixes MervinPraison/PraisonAI#1396) - Add N8nWorkflowTool class for executing n8n workflows - Add n8n_workflow and n8n_list_workflows decorator functions - Provides access to 400+ integrations (Slack, Gmail, Notion, databases, APIs) - Includes comprehensive unit tests with mocking - Add example usage and documentation - Optional httpx dependency with lazy loading - Bidirectional PraisonAI ↔ n8n integration support * fix: resolve critical n8n integration issues - Export n8n_list_workflows from __init__.py (fixes ImportError) - Fix module path in tools registry to use absolute imports - Update __getattr__ to handle absolute module paths - Fix timeout parameter not being used in _wait_for_execution - Update test mocking to work with lazy imports Addresses P0 and P1 issues identified in code review. Co-authored-by: Mervin Praison <[email protected]> --------- Co-authored-by: MervinPraison <[email protected]> Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com> Co-authored-by: Mervin Praison <[email protected]>
This commit is contained in:
7 files changed
+1189
-1
No files matched your search
@@ -0,0 +1,399 @@
|
||||
# n8n Integration for PraisonAI Agents
|
||||
|
||||
Enable bidirectional integration between PraisonAI agents and n8n workflow automation platform, providing access to **400+ integrations** including Slack, Gmail, Notion, databases, APIs, and more.
|
||||
|
||||
## Overview
|
||||
|
||||
The n8n integration allows:
|
||||
|
||||
- **PraisonAI → n8n**: Agents can execute n8n workflows to access 400+ integrations
|
||||
- **n8n → PraisonAI**: n8n workflows can invoke PraisonAI agents (see API endpoint section)
|
||||
|
||||
## Installation
|
||||
|
||||
```bash
|
||||
# Install with n8n support
|
||||
pip install "praisonai-tools[n8n]"
|
||||
```
|
||||
|
||||
## Setup
|
||||
|
||||
### 1. n8n Instance
|
||||
|
||||
**Local Development:**
|
||||
```bash
|
||||
# Run n8n locally with Docker
|
||||
docker run -it --rm --name n8n -p 5678:5678 -v n8n_data:/home/node/.n8n docker.n8n.io/n8nio/n8n
|
||||
```
|
||||
|
||||
**Production:**
|
||||
- Use [n8n Cloud](https://n8n.cloud) (recommended)
|
||||
- Or self-host n8n on your infrastructure
|
||||
|
||||
### 2. Environment Variables
|
||||
|
||||
```bash
|
||||
export N8N_URL="http://localhost:5678" # Your n8n instance URL
|
||||
export N8N_API_KEY="your-api-key" # Optional for local, required for production
|
||||
```
|
||||
|
||||
### 3. Create API Key (Production)
|
||||
|
||||
1. Open n8n instance
|
||||
2. Go to Settings → API
|
||||
3. Generate new API key
|
||||
4. Set `N8N_API_KEY` environment variable
|
||||
|
||||
## Quick Start
|
||||
|
||||
### Basic Usage
|
||||
|
||||
```python
|
||||
from praisonai_tools.n8n import n8n_workflow, n8n_list_workflows
|
||||
|
||||
# List available workflows
|
||||
workflows = n8n_list_workflows()
|
||||
print(f"Available workflows: {len(workflows.get('data', []))}")
|
||||
|
||||
# Execute a workflow
|
||||
result = n8n_workflow(
|
||||
workflow_id="your-workflow-id",
|
||||
input_data={"message": "Hello from PraisonAI!"}
|
||||
)
|
||||
```
|
||||
|
||||
### With PraisonAI Agents
|
||||
|
||||
```python
|
||||
from praisonaiagents import Agent
|
||||
from praisonai_tools.n8n import n8n_workflow, n8n_list_workflows
|
||||
|
||||
# Create agent with n8n tools
|
||||
agent = Agent(
|
||||
name="automation-agent",
|
||||
instructions="You help automate tasks using n8n workflows with 400+ integrations",
|
||||
tools=[n8n_workflow, n8n_list_workflows],
|
||||
)
|
||||
|
||||
# Agent can now use n8n workflows
|
||||
response = agent.start("Send a Slack message to #general saying 'Hello team!'")
|
||||
```
|
||||
|
||||
## Available Tools
|
||||
|
||||
### `n8n_workflow`
|
||||
|
||||
Execute an n8n workflow and return the result.
|
||||
|
||||
**Parameters:**
|
||||
- `workflow_id` (str): The n8n workflow ID to execute
|
||||
- `input_data` (dict, optional): Input data to pass to the workflow
|
||||
- `n8n_url` (str, optional): n8n instance URL (defaults to `N8N_URL` env var)
|
||||
- `api_key` (str, optional): n8n API key (defaults to `N8N_API_KEY` env var)
|
||||
- `timeout` (float, optional): Request timeout in seconds (default: 60)
|
||||
- `wait_for_completion` (bool, optional): Wait for workflow completion (default: True)
|
||||
|
||||
**Returns:** Dict with execution result or error information
|
||||
|
||||
**Example:**
|
||||
```python
|
||||
# Send Slack message via n8n workflow
|
||||
result = n8n_workflow(
|
||||
workflow_id="slack-notify",
|
||||
input_data={
|
||||
"channel": "#general",
|
||||
"message": "Hello from PraisonAI!"
|
||||
}
|
||||
)
|
||||
|
||||
# Create Notion page via n8n workflow
|
||||
result = n8n_workflow(
|
||||
workflow_id="notion-create-page",
|
||||
input_data={
|
||||
"title": "Meeting Notes",
|
||||
"content": "Discussion about project timeline"
|
||||
}
|
||||
)
|
||||
```
|
||||
|
||||
### `n8n_list_workflows`
|
||||
|
||||
List available n8n workflows.
|
||||
|
||||
**Parameters:**
|
||||
- `n8n_url` (str, optional): n8n instance URL
|
||||
- `api_key` (str, optional): n8n API key
|
||||
|
||||
**Returns:** Dict containing list of workflows or error information
|
||||
|
||||
**Example:**
|
||||
```python
|
||||
workflows = n8n_list_workflows()
|
||||
for workflow in workflows.get("data", []):
|
||||
print(f"- {workflow['name']} (ID: {workflow['id']})")
|
||||
```
|
||||
|
||||
## Common Use Cases
|
||||
|
||||
### 1. Communication Automation
|
||||
|
||||
```python
|
||||
# Slack notifications
|
||||
n8n_workflow("slack-notify", {
|
||||
"channel": "#alerts",
|
||||
"message": "Task completed successfully!",
|
||||
"emoji": ":white_check_mark:"
|
||||
})
|
||||
|
||||
# Email with Gmail
|
||||
n8n_workflow("gmail-send", {
|
||||
"to": "[email protected]",
|
||||
"subject": "Report Ready",
|
||||
"body": "Your daily report is attached."
|
||||
})
|
||||
```
|
||||
|
||||
### 2. Data Management
|
||||
|
||||
```python
|
||||
# Add to Google Sheets
|
||||
n8n_workflow("sheets-append", {
|
||||
"spreadsheet_id": "1BxiMVs0XRA5nFMdKvBdBZjgmUUqptlbs74OgvE2upms",
|
||||
"range": "Sheet1!A1:E1",
|
||||
"values": [["Date", "Name", "Status", "Notes"]]
|
||||
})
|
||||
|
||||
# Create Notion page
|
||||
n8n_workflow("notion-create", {
|
||||
"database_id": "your-database-id",
|
||||
"properties": {
|
||||
"Title": {"title": [{"text": {"content": "New Task"}}]},
|
||||
"Status": {"select": {"name": "To Do"}}
|
||||
}
|
||||
})
|
||||
```
|
||||
|
||||
### 3. Task Management
|
||||
|
||||
```python
|
||||
# Create Jira ticket
|
||||
n8n_workflow("jira-create-issue", {
|
||||
"project": "PROJ",
|
||||
"summary": "Bug report from agent",
|
||||
"description": "Issue details...",
|
||||
"issue_type": "Bug"
|
||||
})
|
||||
|
||||
# Add Trello card
|
||||
n8n_workflow("trello-add-card", {
|
||||
"list_id": "your-list-id",
|
||||
"name": "New Task",
|
||||
"desc": "Task description"
|
||||
})
|
||||
```
|
||||
|
||||
## Agent Examples
|
||||
|
||||
### Multi-Platform Notification Agent
|
||||
|
||||
```python
|
||||
agent = Agent(
|
||||
name="notifier",
|
||||
instructions="""
|
||||
You send notifications across multiple platforms using n8n workflows.
|
||||
Available workflows:
|
||||
- slack-notify: Send Slack messages
|
||||
- discord-webhook: Send Discord messages
|
||||
- email-send: Send emails
|
||||
- teams-notify: Send Microsoft Teams messages
|
||||
|
||||
Always ask which platform and customize the message appropriately.
|
||||
""",
|
||||
tools=[n8n_workflow, n8n_list_workflows]
|
||||
)
|
||||
```
|
||||
|
||||
### Data Sync Agent
|
||||
|
||||
```python
|
||||
agent = Agent(
|
||||
name="data-sync",
|
||||
instructions="""
|
||||
You help synchronize data between different systems using n8n.
|
||||
You can:
|
||||
- Sync data between databases
|
||||
- Update spreadsheets
|
||||
- Create/update CRM records
|
||||
- Generate reports
|
||||
|
||||
Always confirm the source and destination before proceeding.
|
||||
""",
|
||||
tools=[n8n_workflow]
|
||||
)
|
||||
```
|
||||
|
||||
## Error Handling
|
||||
|
||||
The n8n tools provide detailed error information:
|
||||
|
||||
```python
|
||||
result = n8n_workflow("invalid-workflow")
|
||||
|
||||
if "error" in result:
|
||||
print(f"Error: {result['error']}")
|
||||
# Common errors:
|
||||
# - "workflow_id is required"
|
||||
# - "httpx not installed. Install with: pip install 'praisonai-tools[n8n]'"
|
||||
# - "HTTP 404: Workflow not found"
|
||||
# - "Workflow execution timed out after 60 seconds"
|
||||
```
|
||||
|
||||
## Configuration
|
||||
|
||||
### Timeout Settings
|
||||
|
||||
```python
|
||||
# Short timeout for quick workflows
|
||||
result = n8n_workflow(
|
||||
workflow_id="quick-task",
|
||||
timeout=10.0
|
||||
)
|
||||
|
||||
# Long timeout for complex workflows
|
||||
result = n8n_workflow(
|
||||
workflow_id="data-processing",
|
||||
timeout=300.0 # 5 minutes
|
||||
)
|
||||
```
|
||||
|
||||
### Async vs Sync Execution
|
||||
|
||||
```python
|
||||
# Wait for completion (default)
|
||||
result = n8n_workflow("workflow-id", wait_for_completion=True)
|
||||
|
||||
# Fire-and-forget
|
||||
result = n8n_workflow("workflow-id", wait_for_completion=False)
|
||||
# Returns immediately with executionId
|
||||
```
|
||||
|
||||
## Integration with n8n → PraisonAI
|
||||
|
||||
To enable n8n workflows to invoke PraisonAI agents, add this endpoint to your PraisonAI application:
|
||||
|
||||
```python
|
||||
from fastapi import FastAPI, HTTPException
|
||||
from pydantic import BaseModel
|
||||
from praisonaiagents import Agent
|
||||
|
||||
app = FastAPI()
|
||||
|
||||
class InvokeRequest(BaseModel):
|
||||
agent_name: str
|
||||
message: str
|
||||
session_id: str = None
|
||||
|
||||
@app.post("/api/v1/agents/{agent_id}/invoke")
|
||||
async def invoke_agent(agent_id: str, request: InvokeRequest):
|
||||
"""Endpoint for n8n to invoke PraisonAI agents."""
|
||||
# Load your agent
|
||||
agent = get_agent(agent_id) # Your agent loading logic
|
||||
if not agent:
|
||||
raise HTTPException(404, f"Agent {agent_id} not found")
|
||||
|
||||
result = await agent.astart(request.message)
|
||||
return {"result": result, "session_id": request.session_id or "default"}
|
||||
```
|
||||
|
||||
Then use n8n's HTTP Request node to call this endpoint.
|
||||
|
||||
## Best Practices
|
||||
|
||||
### 1. Workflow Design
|
||||
- Create focused workflows for specific tasks
|
||||
- Use descriptive workflow names and IDs
|
||||
- Add error handling in your n8n workflows
|
||||
- Test workflows manually before using with agents
|
||||
|
||||
### 2. Security
|
||||
- Use API keys for production environments
|
||||
- Implement input validation in n8n workflows
|
||||
- Consider rate limiting for agent-triggered workflows
|
||||
- Audit workflow executions regularly
|
||||
|
||||
### 3. Performance
|
||||
- Use appropriate timeouts based on workflow complexity
|
||||
- Consider async execution for long-running workflows
|
||||
- Monitor n8n instance performance
|
||||
- Cache workflow lists when possible
|
||||
|
||||
### 4. Error Handling
|
||||
- Always check for error responses
|
||||
- Implement retry logic for transient failures
|
||||
- Log workflow execution results
|
||||
- Provide fallback options when workflows fail
|
||||
|
||||
## Troubleshooting
|
||||
|
||||
### Common Issues
|
||||
|
||||
**Connection Errors:**
|
||||
```python
|
||||
# Check n8n connectivity
|
||||
result = n8n_list_workflows()
|
||||
if "error" in result:
|
||||
print("n8n connection issue:", result["error"])
|
||||
```
|
||||
|
||||
**Workflow Not Found:**
|
||||
```python
|
||||
# Verify workflow exists
|
||||
workflows = n8n_list_workflows()
|
||||
workflow_ids = [w["id"] for w in workflows.get("data", [])]
|
||||
print("Available workflows:", workflow_ids)
|
||||
```
|
||||
|
||||
**Authentication Issues:**
|
||||
```python
|
||||
# Check API key configuration
|
||||
import os
|
||||
api_key = os.getenv("N8N_API_KEY")
|
||||
if not api_key:
|
||||
print("N8N_API_KEY not set - OK for local testing")
|
||||
else:
|
||||
print("API key configured")
|
||||
```
|
||||
|
||||
### Debug Mode
|
||||
|
||||
Enable detailed logging:
|
||||
|
||||
```python
|
||||
import logging
|
||||
logging.basicConfig(level=logging.DEBUG)
|
||||
|
||||
# Now n8n tool calls will show detailed logs
|
||||
result = n8n_workflow("test-workflow")
|
||||
```
|
||||
|
||||
## Resources
|
||||
|
||||
- [n8n Documentation](https://docs.n8n.io/)
|
||||
- [n8n API Reference](https://docs.n8n.io/api/)
|
||||
- [n8n Workflow Templates](https://n8n.io/workflows)
|
||||
- [n8n Integrations](https://n8n.io/integrations)
|
||||
- [PraisonAI Documentation](https://docs.praison.ai)
|
||||
|
||||
## Contributing
|
||||
|
||||
To contribute to the n8n integration:
|
||||
|
||||
1. Fork the [PraisonAI-Tools repository](https://github.com/MervinPraison/PraisonAI-Tools)
|
||||
2. Create your feature branch
|
||||
3. Add tests for your changes
|
||||
4. Update documentation
|
||||
5. Submit a pull request
|
||||
|
||||
## License
|
||||
|
||||
This integration is released under the MIT License, same as PraisonAI Tools.
|
||||
@@ -0,0 +1,182 @@
|
||||
"""n8n Integration Example for PraisonAI Agents.
|
||||
|
||||
This example demonstrates how to use the n8n integration tools to enable
|
||||
PraisonAI agents to access n8n's 400+ integrations (Slack, Gmail, Notion, etc.).
|
||||
|
||||
Prerequisites:
|
||||
1. Install PraisonAI Tools with n8n support:
|
||||
pip install "praisonai-tools[n8n]"
|
||||
|
||||
2. Set up n8n (local or cloud):
|
||||
- Local: docker run -it --rm --name n8n -p 5678:5678 -v n8n_data:/home/node/.n8n docker.n8n.io/n8nio/n8n
|
||||
- Cloud: https://n8n.cloud
|
||||
|
||||
3. Configure environment variables:
|
||||
export N8N_URL="http://localhost:5678" # or your n8n instance URL
|
||||
export N8N_API_KEY="your-api-key" # optional for local testing
|
||||
|
||||
4. Create some workflows in n8n that your agents can use
|
||||
|
||||
Usage:
|
||||
python examples/n8n_integration_example.py
|
||||
"""
|
||||
|
||||
import os
|
||||
from praisonaiagents import Agent
|
||||
from praisonai_tools.n8n import n8n_workflow, n8n_list_workflows
|
||||
|
||||
|
||||
def example_list_workflows():
|
||||
"""Example: List available n8n workflows."""
|
||||
print("🔍 Listing available n8n workflows...")
|
||||
|
||||
try:
|
||||
result = n8n_list_workflows()
|
||||
|
||||
if "error" in result:
|
||||
print(f"❌ Error: {result['error']}")
|
||||
return
|
||||
|
||||
workflows = result.get("data", [])
|
||||
if workflows:
|
||||
print(f"✅ Found {len(workflows)} workflows:")
|
||||
for workflow in workflows[:5]: # Show first 5
|
||||
print(f" - {workflow.get('name', 'Unnamed')} (ID: {workflow.get('id')})")
|
||||
else:
|
||||
print("📝 No workflows found. Create some workflows in n8n first!")
|
||||
|
||||
return workflows
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Exception: {e}")
|
||||
return []
|
||||
|
||||
|
||||
def example_execute_workflow():
|
||||
"""Example: Execute a simple n8n workflow."""
|
||||
print("\n🚀 Executing n8n workflow...")
|
||||
|
||||
# This would be the ID of a workflow you created in n8n
|
||||
# For demo purposes, we'll use a placeholder
|
||||
workflow_id = os.getenv("N8N_TEST_WORKFLOW_ID", "demo-workflow-123")
|
||||
|
||||
try:
|
||||
result = n8n_workflow(
|
||||
workflow_id=workflow_id,
|
||||
input_data={
|
||||
"message": "Hello from PraisonAI!",
|
||||
"timestamp": "2026-04-16T12:00:00Z",
|
||||
"source": "praisonai-agent"
|
||||
},
|
||||
wait_for_completion=True
|
||||
)
|
||||
|
||||
if "error" in result:
|
||||
print(f"❌ Error: {result['error']}")
|
||||
if "workflow" in result["error"].lower():
|
||||
print("💡 Tip: Create a test workflow in n8n and set N8N_TEST_WORKFLOW_ID")
|
||||
return
|
||||
|
||||
print("✅ Workflow executed successfully!")
|
||||
print(f"📊 Result: {result}")
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Exception: {e}")
|
||||
|
||||
|
||||
def example_agent_with_n8n_tools():
|
||||
"""Example: PraisonAI agent using n8n tools."""
|
||||
print("\n🤖 Creating PraisonAI agent with n8n tools...")
|
||||
|
||||
# Create an agent with n8n tools
|
||||
agent = Agent(
|
||||
name="automation-agent",
|
||||
instructions="""
|
||||
You are an automation assistant with access to n8n workflows.
|
||||
You can help users automate tasks using n8n's 400+ integrations.
|
||||
|
||||
Available capabilities through n8n:
|
||||
- Send messages to Slack, Discord, Telegram
|
||||
- Create pages in Notion, Google Docs
|
||||
- Add rows to Google Sheets, Airtable
|
||||
- Send emails via Gmail, Outlook
|
||||
- Create tasks in Trello, Jira, Linear
|
||||
- Store data in databases (PostgreSQL, MongoDB, etc.)
|
||||
- Call external APIs and webhooks
|
||||
|
||||
Always explain what you're doing and ask for workflow IDs when needed.
|
||||
""",
|
||||
tools=[n8n_workflow, n8n_list_workflows],
|
||||
llm="gpt-4o-mini", # Use a lightweight model for demo
|
||||
)
|
||||
|
||||
print("✅ Agent created with n8n tools!")
|
||||
return agent
|
||||
|
||||
|
||||
def example_agent_conversation():
|
||||
"""Example conversation with the n8n-enabled agent."""
|
||||
agent = example_agent_with_n8n_tools()
|
||||
|
||||
if not agent:
|
||||
print("❌ Failed to create agent")
|
||||
return
|
||||
|
||||
print("\n💬 Starting conversation with agent...")
|
||||
|
||||
# Example conversation
|
||||
conversations = [
|
||||
"Can you list the available n8n workflows?",
|
||||
"What can you help me automate using n8n?",
|
||||
# "Execute the Slack notification workflow with the message 'Test from PraisonAI'"
|
||||
]
|
||||
|
||||
for i, message in enumerate(conversations, 1):
|
||||
print(f"\n👤 User #{i}: {message}")
|
||||
try:
|
||||
response = agent.start(message)
|
||||
print(f"🤖 Agent: {response}")
|
||||
except Exception as e:
|
||||
print(f"❌ Error in conversation: {e}")
|
||||
|
||||
|
||||
def main():
|
||||
"""Main function to run all examples."""
|
||||
print("🔗 n8n Integration Example for PraisonAI Agents")
|
||||
print("=" * 50)
|
||||
|
||||
# Check configuration
|
||||
n8n_url = os.getenv("N8N_URL", "http://localhost:5678")
|
||||
n8n_api_key = os.getenv("N8N_API_KEY")
|
||||
|
||||
print(f"🔧 Configuration:")
|
||||
print(f" N8N_URL: {n8n_url}")
|
||||
print(f" N8N_API_KEY: {'Set' if n8n_api_key else 'Not set (OK for local testing)'}")
|
||||
|
||||
# Run examples
|
||||
workflows = example_list_workflows()
|
||||
|
||||
if workflows:
|
||||
example_execute_workflow()
|
||||
else:
|
||||
print("\n💡 To test workflow execution:")
|
||||
print(" 1. Open your n8n instance and create a simple workflow")
|
||||
print(" 2. Set the workflow ID: export N8N_TEST_WORKFLOW_ID='your-workflow-id'")
|
||||
print(" 3. Run this example again")
|
||||
|
||||
# Agent examples (these work even without actual workflows)
|
||||
try:
|
||||
example_agent_conversation()
|
||||
except Exception as e:
|
||||
print(f"❌ Agent example failed: {e}")
|
||||
print("💡 This might be due to missing OpenAI API key or n8n connectivity")
|
||||
|
||||
print("\n✨ Example completed!")
|
||||
print("\n📚 Next steps:")
|
||||
print(" - Create workflows in your n8n instance")
|
||||
print(" - Use the agent in your own applications")
|
||||
print(" - Explore n8n's 400+ integrations at https://n8n.io/integrations")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,23 @@
|
||||
"""n8n Workflow Automation Tools for PraisonAI Agents.
|
||||
|
||||
Bidirectional integration between PraisonAI agents and n8n workflow automation platform.
|
||||
|
||||
Usage:
|
||||
from praisonai_tools.n8n import n8n_workflow
|
||||
|
||||
# Agent uses n8n workflow
|
||||
result = n8n_workflow(
|
||||
workflow_id="my-workflow-id",
|
||||
input_data={"message": "Hello from PraisonAI"},
|
||||
n8n_url="http://localhost:5678",
|
||||
api_key="your-n8n-api-key"
|
||||
)
|
||||
"""
|
||||
|
||||
from .n8n_workflow import n8n_workflow, n8n_list_workflows, N8nWorkflowTool
|
||||
|
||||
__all__ = [
|
||||
"n8n_workflow",
|
||||
"n8n_list_workflows",
|
||||
"N8nWorkflowTool",
|
||||
]
|
||||
@@ -0,0 +1,252 @@
|
||||
"""n8n Workflow Execution Tool for PraisonAI Agents.
|
||||
|
||||
Enables PraisonAI agents to execute n8n workflows, providing access to
|
||||
n8n's 400+ integrations (Slack, Gmail, Notion, databases, APIs, etc.).
|
||||
|
||||
Usage:
|
||||
from praisonai_tools.n8n import n8n_workflow
|
||||
|
||||
# Execute n8n workflow
|
||||
result = n8n_workflow(
|
||||
workflow_id="workflow-123",
|
||||
input_data={"message": "Hello from PraisonAI"},
|
||||
n8n_url="http://localhost:5678",
|
||||
api_key="your-api-key"
|
||||
)
|
||||
|
||||
Environment Variables:
|
||||
N8N_URL: n8n instance URL (default: http://localhost:5678)
|
||||
N8N_API_KEY: n8n API key for authentication
|
||||
"""
|
||||
|
||||
import os
|
||||
import logging
|
||||
from typing import Any, Dict, Optional
|
||||
|
||||
from praisonai_tools.tools.base import BaseTool
|
||||
from praisonai_tools.tools.decorator import tool
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class N8nWorkflowTool(BaseTool):
|
||||
"""Tool for executing n8n workflows from PraisonAI agents."""
|
||||
|
||||
name = "n8n_workflow"
|
||||
description = "Execute n8n workflows to access 400+ integrations (Slack, Gmail, Notion, databases, APIs, etc.)"
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
n8n_url: Optional[str] = None,
|
||||
api_key: Optional[str] = None,
|
||||
timeout: float = 60.0,
|
||||
):
|
||||
self.n8n_url = (n8n_url or os.getenv("N8N_URL", "http://localhost:5678")).rstrip('/')
|
||||
self.api_key = api_key or os.getenv("N8N_API_KEY")
|
||||
self.timeout = timeout
|
||||
super().__init__()
|
||||
|
||||
def run(
|
||||
self,
|
||||
workflow_id: str,
|
||||
input_data: Optional[Dict[str, Any]] = None,
|
||||
wait_for_completion: bool = True,
|
||||
**kwargs
|
||||
) -> Dict[str, Any]:
|
||||
"""Execute an n8n workflow and return the result.
|
||||
|
||||
Args:
|
||||
workflow_id: The n8n workflow ID to execute
|
||||
input_data: Input data to pass to the workflow
|
||||
wait_for_completion: Whether to wait for workflow completion
|
||||
|
||||
Returns:
|
||||
Workflow execution result
|
||||
"""
|
||||
if not workflow_id:
|
||||
return {"error": "workflow_id is required"}
|
||||
|
||||
try:
|
||||
# Lazy import httpx to avoid import-time overhead
|
||||
import httpx
|
||||
except ImportError:
|
||||
return {"error": "httpx not installed. Install with: pip install 'praisonai-tools[n8n]'"}
|
||||
|
||||
# Prepare headers
|
||||
headers = {"Content-Type": "application/json"}
|
||||
if self.api_key:
|
||||
headers["X-N8N-API-KEY"] = self.api_key
|
||||
|
||||
# Execute workflow
|
||||
try:
|
||||
with httpx.Client(timeout=self.timeout) as client:
|
||||
response = client.post(
|
||||
f"{self.n8n_url}/api/v1/workflows/{workflow_id}/execute",
|
||||
json={"data": input_data or {}},
|
||||
headers=headers,
|
||||
)
|
||||
response.raise_for_status()
|
||||
|
||||
result = response.json()
|
||||
|
||||
if wait_for_completion and result.get("executionId"):
|
||||
# Poll for completion
|
||||
execution_id = result["executionId"]
|
||||
return self._wait_for_execution(client, execution_id, headers)
|
||||
|
||||
return result
|
||||
|
||||
except httpx.TimeoutException:
|
||||
logger.error(f"n8n workflow {workflow_id} timed out after {self.timeout}s")
|
||||
return {"error": f"Workflow execution timed out after {self.timeout} seconds"}
|
||||
except httpx.HTTPStatusError as e:
|
||||
logger.error(f"n8n API error: {e.response.status_code} - {e.response.text}")
|
||||
return {"error": f"HTTP {e.response.status_code}: {e.response.text}"}
|
||||
except Exception as e:
|
||||
logger.error(f"n8n workflow execution error: {e}")
|
||||
return {"error": str(e)}
|
||||
|
||||
def _wait_for_execution(
|
||||
self,
|
||||
client: "httpx.Client",
|
||||
execution_id: str,
|
||||
headers: Dict[str, str],
|
||||
max_wait: Optional[int] = None,
|
||||
poll_interval: int = 2,
|
||||
) -> Dict[str, Any]:
|
||||
"""Wait for workflow execution to complete."""
|
||||
import time
|
||||
|
||||
# Use configured timeout if max_wait is not provided
|
||||
if max_wait is None:
|
||||
max_wait = int(self.timeout)
|
||||
|
||||
waited = 0
|
||||
while waited < max_wait:
|
||||
try:
|
||||
response = client.get(
|
||||
f"{self.n8n_url}/api/v1/executions/{execution_id}",
|
||||
headers=headers,
|
||||
)
|
||||
response.raise_for_status()
|
||||
|
||||
execution = response.json()
|
||||
status = execution.get("status")
|
||||
|
||||
if status in ["success", "error", "canceled"]:
|
||||
return execution
|
||||
|
||||
time.sleep(poll_interval)
|
||||
waited += poll_interval
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error polling execution {execution_id}: {e}")
|
||||
return {"error": f"Error polling execution: {e}"}
|
||||
|
||||
return {"error": f"Execution {execution_id} did not complete within {max_wait} seconds"}
|
||||
|
||||
def list_workflows(self) -> Dict[str, Any]:
|
||||
"""List available n8n workflows."""
|
||||
try:
|
||||
import httpx
|
||||
except ImportError:
|
||||
return {"error": "httpx not installed. Install with: pip install 'praisonai-tools[n8n]'"}
|
||||
|
||||
headers = {"Content-Type": "application/json"}
|
||||
if self.api_key:
|
||||
headers["X-N8N-API-KEY"] = self.api_key
|
||||
|
||||
try:
|
||||
with httpx.Client(timeout=30.0) as client:
|
||||
response = client.get(
|
||||
f"{self.n8n_url}/api/v1/workflows",
|
||||
headers=headers,
|
||||
)
|
||||
response.raise_for_status()
|
||||
return response.json()
|
||||
|
||||
except httpx.HTTPStatusError as e:
|
||||
logger.error(f"n8n API error: {e.response.status_code} - {e.response.text}")
|
||||
return {"error": f"HTTP {e.response.status_code}: {e.response.text}"}
|
||||
except Exception as e:
|
||||
logger.error(f"n8n list workflows error: {e}")
|
||||
return {"error": str(e)}
|
||||
|
||||
|
||||
@tool
|
||||
def n8n_workflow(
|
||||
workflow_id: str,
|
||||
input_data: Optional[Dict[str, Any]] = None,
|
||||
n8n_url: Optional[str] = None,
|
||||
api_key: Optional[str] = None,
|
||||
timeout: float = 60.0,
|
||||
wait_for_completion: bool = True,
|
||||
) -> Dict[str, Any]:
|
||||
"""Execute an n8n workflow and return the result.
|
||||
|
||||
Provides PraisonAI agents access to n8n's 400+ integrations including:
|
||||
- Communication: Slack, Discord, Telegram, Gmail, Outlook
|
||||
- Productivity: Notion, Google Sheets, Airtable, Trello
|
||||
- Databases: PostgreSQL, MongoDB, MySQL, Redis
|
||||
- APIs: REST, GraphQL, webhooks
|
||||
- And 400+ more integrations
|
||||
|
||||
Args:
|
||||
workflow_id: The n8n workflow ID to execute
|
||||
input_data: Input data to pass to the workflow (optional)
|
||||
n8n_url: n8n instance URL (defaults to N8N_URL env var or http://localhost:5678)
|
||||
api_key: n8n API key (defaults to N8N_API_KEY env var)
|
||||
timeout: Request timeout in seconds (default: 60)
|
||||
wait_for_completion: Wait for workflow to complete (default: True)
|
||||
|
||||
Returns:
|
||||
Dict containing workflow execution result or error information
|
||||
|
||||
Example:
|
||||
# Send Slack message via n8n workflow
|
||||
result = n8n_workflow(
|
||||
workflow_id="slack-notify",
|
||||
input_data={
|
||||
"channel": "#general",
|
||||
"message": "Hello from PraisonAI!"
|
||||
}
|
||||
)
|
||||
|
||||
# Create Notion page via n8n workflow
|
||||
result = n8n_workflow(
|
||||
workflow_id="notion-create-page",
|
||||
input_data={
|
||||
"title": "Meeting Notes",
|
||||
"content": "Discussion about project timeline"
|
||||
}
|
||||
)
|
||||
"""
|
||||
tool = N8nWorkflowTool(
|
||||
n8n_url=n8n_url,
|
||||
api_key=api_key,
|
||||
timeout=timeout,
|
||||
)
|
||||
|
||||
return tool.run(
|
||||
workflow_id=workflow_id,
|
||||
input_data=input_data,
|
||||
wait_for_completion=wait_for_completion,
|
||||
)
|
||||
|
||||
|
||||
@tool
|
||||
def n8n_list_workflows(
|
||||
n8n_url: Optional[str] = None,
|
||||
api_key: Optional[str] = None,
|
||||
) -> Dict[str, Any]:
|
||||
"""List available n8n workflows.
|
||||
|
||||
Args:
|
||||
n8n_url: n8n instance URL (defaults to N8N_URL env var or http://localhost:5678)
|
||||
api_key: n8n API key (defaults to N8N_API_KEY env var)
|
||||
|
||||
Returns:
|
||||
Dict containing list of available workflows or error information
|
||||
"""
|
||||
tool = N8nWorkflowTool(n8n_url=n8n_url, api_key=api_key)
|
||||
return tool.list_workflows()
|
||||
@@ -529,12 +529,23 @@ def __getattr__(name):
|
||||
"medium_get_user": "medium_tool",
|
||||
"medium_list_publications": "medium_tool",
|
||||
"medium_publish_to_publication": "medium_tool",
|
||||
# n8n Tools
|
||||
"N8nWorkflowTool": "praisonai_tools.n8n.n8n_workflow",
|
||||
"n8n_workflow": "praisonai_tools.n8n.n8n_workflow",
|
||||
"n8n_list_workflows": "praisonai_tools.n8n.n8n_workflow",
|
||||
}
|
||||
|
||||
if name in tool_map:
|
||||
module_name = tool_map[name]
|
||||
from importlib import import_module
|
||||
module = import_module(f".{module_name}", __package__)
|
||||
|
||||
# Handle absolute module paths (e.g., praisonai_tools.n8n.n8n_workflow)
|
||||
if module_name.startswith("praisonai_tools."):
|
||||
module = import_module(module_name)
|
||||
else:
|
||||
# Handle relative module paths within tools package
|
||||
module = import_module(f".{module_name}", __package__)
|
||||
|
||||
return getattr(module, name)
|
||||
|
||||
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
|
||||
@@ -859,4 +870,8 @@ __all__ = [
|
||||
"medium_get_user",
|
||||
"medium_list_publications",
|
||||
"medium_publish_to_publication",
|
||||
# n8n Tool
|
||||
"N8nWorkflowTool",
|
||||
"n8n_workflow",
|
||||
"n8n_list_workflows",
|
||||
]
|
||||
@@ -51,6 +51,9 @@ joy-trust = [
|
||||
agentfolio = [
|
||||
"httpx>=0.24.0",
|
||||
]
|
||||
n8n = [
|
||||
"httpx>=0.27.0",
|
||||
]
|
||||
|
||||
[project.urls]
|
||||
Homepage = "https://docs.praison.ai"
|
||||
|
||||
@@ -0,0 +1,314 @@
|
||||
"""Unit tests for n8n integration tools."""
|
||||
|
||||
import os
|
||||
import pytest
|
||||
from unittest.mock import Mock, patch, MagicMock
|
||||
|
||||
|
||||
class TestN8nWorkflowTool:
|
||||
"""Test the N8nWorkflowTool class."""
|
||||
|
||||
@pytest.fixture
|
||||
def mock_httpx(self):
|
||||
# Mock httpx in sys.modules to catch lazy imports
|
||||
with patch.dict('sys.modules', {'httpx': MagicMock()}) as mock_modules:
|
||||
mock_httpx = mock_modules['httpx']
|
||||
yield mock_httpx
|
||||
|
||||
def test_import_n8n_tools(self):
|
||||
"""Test that n8n tools can be imported."""
|
||||
try:
|
||||
from praisonai_tools.n8n import n8n_workflow, N8nWorkflowTool
|
||||
assert callable(n8n_workflow)
|
||||
assert N8nWorkflowTool is not None
|
||||
except ImportError as e:
|
||||
pytest.fail(f"Failed to import n8n tools: {e}")
|
||||
|
||||
def test_n8n_workflow_tool_init(self):
|
||||
"""Test N8nWorkflowTool initialization."""
|
||||
from praisonai_tools.n8n import N8nWorkflowTool
|
||||
|
||||
# Test with default values
|
||||
tool = N8nWorkflowTool()
|
||||
assert tool.n8n_url == "http://localhost:5678"
|
||||
assert tool.api_key is None
|
||||
assert tool.timeout == 60.0
|
||||
|
||||
# Test with custom values
|
||||
tool = N8nWorkflowTool(
|
||||
n8n_url="https://my-n8n.com",
|
||||
api_key="test-key",
|
||||
timeout=30.0
|
||||
)
|
||||
assert tool.n8n_url == "https://my-n8n.com"
|
||||
assert tool.api_key == "test-key"
|
||||
assert tool.timeout == 30.0
|
||||
|
||||
def test_n8n_workflow_tool_env_vars(self):
|
||||
"""Test N8nWorkflowTool reads environment variables."""
|
||||
from praisonai_tools.n8n import N8nWorkflowTool
|
||||
|
||||
with patch.dict(os.environ, {
|
||||
'N8N_URL': 'https://env-n8n.com',
|
||||
'N8N_API_KEY': 'env-api-key'
|
||||
}):
|
||||
tool = N8nWorkflowTool()
|
||||
assert tool.n8n_url == "https://env-n8n.com"
|
||||
assert tool.api_key == "env-api-key"
|
||||
|
||||
def test_n8n_workflow_missing_workflow_id(self):
|
||||
"""Test error handling when workflow_id is missing."""
|
||||
from praisonai_tools.n8n import N8nWorkflowTool
|
||||
|
||||
tool = N8nWorkflowTool()
|
||||
result = tool.run(workflow_id="")
|
||||
assert result["error"] == "workflow_id is required"
|
||||
|
||||
def test_n8n_workflow_missing_httpx(self):
|
||||
"""Test error handling when httpx is not installed."""
|
||||
from praisonai_tools.n8n import N8nWorkflowTool
|
||||
|
||||
tool = N8nWorkflowTool()
|
||||
with patch('praisonai_tools.n8n.n8n_workflow.httpx', None):
|
||||
with patch('builtins.__import__', side_effect=ImportError("No module named 'httpx'")):
|
||||
result = tool.run(workflow_id="test-workflow")
|
||||
assert "httpx not installed" in result["error"]
|
||||
assert "pip install 'praisonai-tools[n8n]'" in result["error"]
|
||||
|
||||
def test_n8n_workflow_successful_execution(self, mock_httpx):
|
||||
"""Test successful workflow execution."""
|
||||
from praisonai_tools.n8n import N8nWorkflowTool
|
||||
|
||||
# Mock httpx response
|
||||
mock_response = Mock()
|
||||
mock_response.json.return_value = {
|
||||
"executionId": "exec-123",
|
||||
"status": "running"
|
||||
}
|
||||
mock_response.raise_for_status.return_value = None
|
||||
|
||||
mock_client = Mock()
|
||||
mock_client.post.return_value = mock_response
|
||||
mock_httpx.Client.return_value.__enter__.return_value = mock_client
|
||||
|
||||
tool = N8nWorkflowTool(api_key="test-key")
|
||||
result = tool.run(
|
||||
workflow_id="test-workflow",
|
||||
input_data={"message": "Hello"},
|
||||
wait_for_completion=False
|
||||
)
|
||||
|
||||
assert result["executionId"] == "exec-123"
|
||||
assert result["status"] == "running"
|
||||
|
||||
# Verify API call
|
||||
mock_client.post.assert_called_once_with(
|
||||
"http://localhost:5678/api/v1/workflows/test-workflow/execute",
|
||||
json={"data": {"message": "Hello"}},
|
||||
headers={"Content-Type": "application/json", "X-N8N-API-KEY": "test-key"},
|
||||
)
|
||||
|
||||
def test_n8n_workflow_http_error(self, mock_httpx):
|
||||
"""Test HTTP error handling."""
|
||||
from praisonai_tools.n8n import N8nWorkflowTool
|
||||
|
||||
# Mock HTTP error
|
||||
mock_response = Mock()
|
||||
mock_response.status_code = 401
|
||||
mock_response.text = "Unauthorized"
|
||||
|
||||
mock_client = Mock()
|
||||
mock_client.post.side_effect = mock_httpx.HTTPStatusError(
|
||||
"401 Unauthorized", request=Mock(), response=mock_response
|
||||
)
|
||||
mock_httpx.Client.return_value.__enter__.return_value = mock_client
|
||||
|
||||
tool = N8nWorkflowTool(api_key="invalid-key")
|
||||
result = tool.run(workflow_id="test-workflow")
|
||||
|
||||
assert "HTTP 401: Unauthorized" in result["error"]
|
||||
|
||||
def test_n8n_workflow_timeout_error(self, mock_httpx):
|
||||
"""Test timeout error handling."""
|
||||
from praisonai_tools.n8n import N8nWorkflowTool
|
||||
|
||||
mock_client = Mock()
|
||||
mock_client.post.side_effect = mock_httpx.TimeoutException("Request timed out")
|
||||
mock_httpx.Client.return_value.__enter__.return_value = mock_client
|
||||
|
||||
tool = N8nWorkflowTool(timeout=5.0)
|
||||
result = tool.run(workflow_id="test-workflow")
|
||||
|
||||
assert "timed out after 5.0 seconds" in result["error"]
|
||||
|
||||
def test_n8n_list_workflows(self, mock_httpx):
|
||||
"""Test listing workflows."""
|
||||
from praisonai_tools.n8n import N8nWorkflowTool
|
||||
|
||||
# Mock httpx response
|
||||
mock_response = Mock()
|
||||
mock_response.json.return_value = {
|
||||
"data": [
|
||||
{"id": "workflow-1", "name": "My Workflow 1"},
|
||||
{"id": "workflow-2", "name": "My Workflow 2"},
|
||||
]
|
||||
}
|
||||
mock_response.raise_for_status.return_value = None
|
||||
|
||||
mock_client = Mock()
|
||||
mock_client.get.return_value = mock_response
|
||||
mock_httpx.Client.return_value.__enter__.return_value = mock_client
|
||||
|
||||
tool = N8nWorkflowTool(api_key="test-key")
|
||||
result = tool.list_workflows()
|
||||
|
||||
assert "data" in result
|
||||
assert len(result["data"]) == 2
|
||||
assert result["data"][0]["name"] == "My Workflow 1"
|
||||
|
||||
# Verify API call
|
||||
mock_client.get.assert_called_once_with(
|
||||
"http://localhost:5678/api/v1/workflows",
|
||||
headers={"Content-Type": "application/json", "X-N8N-API-KEY": "test-key"},
|
||||
)
|
||||
|
||||
|
||||
class TestN8nWorkflowFunction:
|
||||
"""Test the n8n_workflow function."""
|
||||
|
||||
@patch('praisonai_tools.n8n.n8n_workflow.N8nWorkflowTool')
|
||||
def test_n8n_workflow_function_calls_tool(self, mock_tool_class):
|
||||
"""Test that n8n_workflow function creates tool and calls run method."""
|
||||
from praisonai_tools.n8n import n8n_workflow
|
||||
|
||||
# Mock tool instance
|
||||
mock_tool = Mock()
|
||||
mock_tool.run.return_value = {"result": "success"}
|
||||
mock_tool_class.return_value = mock_tool
|
||||
|
||||
result = n8n_workflow(
|
||||
workflow_id="test-workflow",
|
||||
input_data={"key": "value"},
|
||||
n8n_url="https://custom.n8n.com",
|
||||
api_key="custom-key",
|
||||
timeout=30.0
|
||||
)
|
||||
|
||||
# Verify tool was created with correct parameters
|
||||
mock_tool_class.assert_called_once_with(
|
||||
n8n_url="https://custom.n8n.com",
|
||||
api_key="custom-key",
|
||||
timeout=30.0
|
||||
)
|
||||
|
||||
# Verify run was called with correct parameters
|
||||
mock_tool.run.assert_called_once_with(
|
||||
workflow_id="test-workflow",
|
||||
input_data={"key": "value"},
|
||||
wait_for_completion=True
|
||||
)
|
||||
|
||||
assert result == {"result": "success"}
|
||||
|
||||
|
||||
class TestN8nListWorkflowsFunction:
|
||||
"""Test the n8n_list_workflows function."""
|
||||
|
||||
@patch('praisonai_tools.n8n.n8n_workflow.N8nWorkflowTool')
|
||||
def test_n8n_list_workflows_function(self, mock_tool_class):
|
||||
"""Test that n8n_list_workflows function works correctly."""
|
||||
from praisonai_tools.n8n import n8n_list_workflows
|
||||
|
||||
# Mock tool instance
|
||||
mock_tool = Mock()
|
||||
mock_tool.list_workflows.return_value = {"workflows": ["workflow-1"]}
|
||||
mock_tool_class.return_value = mock_tool
|
||||
|
||||
result = n8n_list_workflows(
|
||||
n8n_url="https://test.n8n.com",
|
||||
api_key="test-key"
|
||||
)
|
||||
|
||||
# Verify tool was created correctly
|
||||
mock_tool_class.assert_called_once_with(
|
||||
n8n_url="https://test.n8n.com",
|
||||
api_key="test-key"
|
||||
)
|
||||
|
||||
# Verify list_workflows was called
|
||||
mock_tool.list_workflows.assert_called_once()
|
||||
|
||||
assert result == {"workflows": ["workflow-1"]}
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
class TestN8nIntegration:
|
||||
"""Integration tests for n8n tools (requires running n8n instance)."""
|
||||
|
||||
@pytest.fixture
|
||||
def n8n_config(self):
|
||||
"""Get n8n configuration from environment or skip test."""
|
||||
n8n_url = os.getenv("N8N_URL")
|
||||
if not n8n_url:
|
||||
pytest.skip("N8N_URL environment variable not set")
|
||||
|
||||
return {
|
||||
"n8n_url": n8n_url,
|
||||
"api_key": os.getenv("N8N_API_KEY"), # Optional for local testing
|
||||
}
|
||||
|
||||
def test_list_workflows_integration(self, n8n_config):
|
||||
"""Test listing workflows against real n8n instance."""
|
||||
from praisonai_tools.n8n import n8n_list_workflows
|
||||
|
||||
result = n8n_list_workflows(**n8n_config)
|
||||
|
||||
# Should not have an error key if successful
|
||||
assert "error" not in result
|
||||
# Should return some data structure
|
||||
assert isinstance(result, dict)
|
||||
|
||||
def test_workflow_execution_integration(self, n8n_config):
|
||||
"""Test workflow execution against real n8n instance."""
|
||||
from praisonai_tools.n8n import n8n_workflow
|
||||
|
||||
# Skip if no test workflow ID provided
|
||||
test_workflow_id = os.getenv("N8N_TEST_WORKFLOW_ID")
|
||||
if not test_workflow_id:
|
||||
pytest.skip("N8N_TEST_WORKFLOW_ID environment variable not set")
|
||||
|
||||
result = n8n_workflow(
|
||||
workflow_id=test_workflow_id,
|
||||
input_data={"test": "integration"},
|
||||
**n8n_config
|
||||
)
|
||||
|
||||
# Should not have an error key if successful
|
||||
assert "error" not in result
|
||||
# Should return some execution data
|
||||
assert isinstance(result, dict)
|
||||
|
||||
|
||||
def test_n8n_smoke_test():
|
||||
"""Smoke test to verify n8n tools can be imported without errors."""
|
||||
try:
|
||||
from praisonai_tools.n8n import n8n_workflow, n8n_list_workflows, N8nWorkflowTool
|
||||
|
||||
# Test that they are callable/instantiable
|
||||
assert callable(n8n_workflow)
|
||||
assert callable(n8n_list_workflows)
|
||||
assert N8nWorkflowTool is not None
|
||||
|
||||
# Test basic instantiation
|
||||
tool = N8nWorkflowTool()
|
||||
assert tool.name == "n8n_workflow"
|
||||
assert "n8n" in tool.description.lower()
|
||||
|
||||
except Exception as e:
|
||||
pytest.fail(f"n8n tools smoke test failed: {e}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
# Run smoke test when executed directly
|
||||
test_n8n_smoke_test()
|
||||
print("✅ n8n tools smoke test passed")
|
||||
Reference in new issue
Block a user