import requests
import json
def create_custom_workflow(api_key, messages, repo_url, github_token, branch="main", working_dir=None):
"""
Execute a custom message workflow.
"""
url = "https://api.cloudcoding.ai/invoke"
headers = {
"Content-Type": "application/json",
"x-api-key": api_key
}
payload = {
"messages": messages,
"repo_url": repo_url,
"github_token": github_token,
"branch": branch,
"haiku": True
}
# Add workingDir to the first message if specified
if working_dir and messages:
messages[0]["workingDir"] = working_dir
response = requests.post(url, headers=headers, json=payload, stream=True)
for line in response.iter_lines():
if line.startswith(b'data: '):
try:
data = json.loads(line[6:])
yield data
except json.JSONDecodeError:
continue
def multi_directory_workflow(api_key, workflow_steps, repo_url, github_token, branch="main"):
"""
Execute a workflow across multiple directories using separate API calls.
Each step is a dict with 'directory', 'messages', and optional 'description'.
"""
results = []
for i, step in enumerate(workflow_steps):
print(f"Executing step {i+1}: {step.get('description', f'Work in {step[\"directory\"]}')}")
# Each directory requires a separate API call
messages = step["messages"]
if step["directory"]:
# Add workingDir to the first message
if messages and isinstance(messages[0], dict):
messages[0]["workingDir"] = step["directory"]
step_results = list(create_custom_workflow(
api_key, messages, repo_url, github_token, branch
))
results.append({
"step": i+1,
"directory": step["directory"],
"results": step_results
})
return results
# Example 1: Single directory workflow
single_dir_messages = [
{
"tools": "all",
"content": "Add comprehensive error handling and logging to this application"
},
{
"tools": ["Bash"],
"content": "Run tests to ensure error handling works correctly",
"continueConversation": True
}
]
print("Single directory workflow:")
for event in create_custom_workflow("your_api_key", single_dir_messages,
"https://github.com/user/repo", "ghp_token"):
print(f"Event: {event}")
# Example 2: Multi-directory workflow (separate API calls)
workflow_steps = [
{
"directory": "./backend",
"description": "Backend API improvements",
"messages": [
{
"tools": ["Read", "Edit"],
"content": "Improve API error handling and add request validation"
}
]
},
{
"directory": "./frontend",
"description": "Frontend error handling",
"messages": [
{
"tools": ["Edit", "Write"],
"content": "Add error boundaries and improved error messaging to React components"
}
]
},
{
"directory": None, # Repository root
"description": "Integration testing",
"messages": [
{
"tools": ["Bash", "Write"],
"content": "Run end-to-end tests to validate error handling across the full stack"
}
]
}
]
print("\nMulti-directory workflow:")
results = multi_directory_workflow("your_api_key", workflow_steps,
"https://github.com/user/repo", "ghp_token",
branch="error-handling-improvements")
for result in results:
print(f"Step {result['step']} ({'Root' if not result['directory'] else result['directory']}): {len(result['results'])} events")