Loading...
Loading...
API integration patterns for autonomous-dev including subprocess safety, GitHub CLI integration, retry logic, authentication, rate limiting, and timeout handling. Use when integrating external APIs or CLI tools.
npx skill4agent add akaszubski/autonomous-dev api-integration-patterns["gh", "issue", "create"]import subprocess
from typing import List
def safe_subprocess(
command: List[str],
*,
allowed_commands: List[str],
timeout: int = 30
) -> subprocess.CompletedProcess:
"""Execute subprocess with CWE-78 prevention.
Args:
command: Command and arguments as list (NOT string!)
allowed_commands: Whitelist of allowed commands
timeout: Maximum execution time in seconds
Returns:
Completed subprocess result
Raises:
SecurityError: If command not in whitelist
subprocess.TimeoutExpired: If timeout exceeded
Security:
- CWE-78 Prevention: Argument arrays (no shell injection)
- Command Whitelist: Only approved commands
- Timeout: DoS prevention
Example:
>>> result = safe_subprocess(
... ["gh", "issue", "create", "--title", user_title],
... allowed_commands=["gh", "git"]
... )
"""
# Whitelist validation
if command[0] not in allowed_commands:
raise SecurityError(f"Command not allowed: {command[0]}")
# Execute with argument array (NEVER shell=True!)
return subprocess.run(
command,
capture_output=True,
text=True,
timeout=timeout,
check=True,
shell=False # CRITICAL
)docs/subprocess-safety.mdexamples/safe-subprocess-example.pydef create_github_issue(
title: str,
body: str,
*,
labels: Optional[List[str]] = None,
timeout: int = 30
) -> str:
"""Create GitHub issue using gh CLI.
Args:
title: Issue title
body: Issue body (markdown)
labels: Issue labels (default: None)
timeout: Command timeout in seconds
Returns:
Issue URL
Raises:
subprocess.CalledProcessError: If gh command fails
RuntimeError: If gh CLI not installed
Example:
>>> url = create_github_issue(
... "Bug: Login fails",
... "Login button doesn't work",
... labels=["bug", "p1"]
... )
"""
# Build gh command (argument array)
cmd = ["gh", "issue", "create", "--title", title, "--body", body]
if labels:
for label in labels:
cmd.extend(["--label", label])
# Execute safely
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=timeout,
check=True,
shell=False
)
# Extract URL from output
return result.stdout.strip()docs/github-cli-integration.mdexamples/github-issue-example.pyimport time
from typing import Callable, TypeVar, Any
T = TypeVar('T')
def retry_with_backoff(
func: Callable[..., T],
*,
max_attempts: int = 3,
base_delay: float = 1.0,
max_delay: float = 60.0
) -> T:
"""Retry function with exponential backoff.
Args:
func: Function to retry
max_attempts: Maximum retry attempts
base_delay: Initial delay in seconds
max_delay: Maximum delay in seconds
Returns:
Function result
Raises:
Exception: Last exception if all retries fail
Example:
>>> result = retry_with_backoff(
... lambda: api_call(),
... max_attempts=5,
... base_delay=2.0
... )
"""
last_exception = None
for attempt in range(max_attempts):
try:
return func()
except Exception as e:
last_exception = e
if attempt < max_attempts - 1:
# Exponential backoff: 1s, 2s, 4s, 8s, ...
delay = min(base_delay * (2 ** attempt), max_delay)
time.sleep(delay)
raise last_exceptiondocs/retry-logic.mdtemplates/retry-decorator-template.pyimport os
from typing import Optional
def get_github_token() -> str:
"""Get GitHub token from environment.
Returns:
GitHub personal access token
Raises:
RuntimeError: If token not found
Security:
- Environment Variables: Never hardcode tokens
- Validation: Check token format
- No Logging: Never log credentials
"""
token = os.getenv("GITHUB_TOKEN")
if not token:
raise RuntimeError(
"GITHUB_TOKEN not found in environment\n"
"Set with: export GITHUB_TOKEN=your_token\n"
"Or add to .env file"
)
# Validate token format (basic check)
if not token.startswith("ghp_") and not token.startswith("github_pat_"):
raise ValueError("Invalid GitHub token format")
return tokendocs/authentication-patterns.mdtemplates/github-api-template.pyimport time
from datetime import datetime, timedelta
class RateLimiter:
"""Simple rate limiter for API calls.
Attributes:
max_calls: Maximum calls per window
window_seconds: Time window in seconds
"""
def __init__(self, max_calls: int, window_seconds: int):
self.max_calls = max_calls
self.window_seconds = window_seconds
self.calls = []
def wait_if_needed(self) -> None:
"""Wait if rate limit would be exceeded."""
now = datetime.now()
cutoff = now - timedelta(seconds=self.window_seconds)
# Remove old calls outside window
self.calls = [c for c in self.calls if c > cutoff]
# Wait if at limit
if len(self.calls) >= self.max_calls:
oldest = self.calls[0]
wait_until = oldest + timedelta(seconds=self.window_seconds)
wait_seconds = (wait_until - now).total_seconds()
if wait_seconds > 0:
time.sleep(wait_seconds)
# Retry removal after wait
self.calls = [c for c in self.calls if c > cutoff]
# Record this call
self.calls.append(now)docs/rate-limiting.mdexamples/github-api-example.pytemplates/subprocess-executor-template.pytemplates/retry-decorator-template.pytemplates/github-api-template.pyexamples/github-issue-example.pyexamples/github-pr-example.pyexamples/safe-subprocess-example.pydocs/subprocess-safety.mddocs/github-cli-integration.mddocs/retry-logic.mddocs/authentication-patterns.md