Python Orchestration

Python orchestration wraps Gemini CLI in subprocess calls, gaining access to Python's rich ecosystem for data processing, file manipulation, API integration, and error handling. The subprocess.run() function executes gemini -p, captures stdout and stderr, and returns the exit code. Python then parses the JSON output, applies business logic, and routes results to databases, APIs, dashboards, or downstream scripts. This pattern is ideal when your automation needs data transformation that is awkward in bash — JSON manipulation, CSV processing, HTTP requests, or database writes.

The async variant uses asyncio.create_subprocess_exec() for concurrent Gemini invocations. Python's asyncio event loop manages multiple concurrent subprocess calls without threads, with fine-grained control over concurrency via asyncio.Semaphore. This is more flexible than xargs -P or GNU parallel for complex orchestration — you can implement adaptive rate limiting, conditional branching based on results, and aggregation of concurrent outputs in pure Python.

import subprocess
import json
import asyncio
from pathlib import Path

def gemini_analyze(file_path: str, prompt: str) -> dict | None:
    """Run gemini -p and return parsed JSON response."""
    full_prompt = f"{prompt}\n\n{Path(file_path).read_text()}"
    result = subprocess.run(
        ["gemini", "-p", full_prompt, "--output-format", "json"],
        capture_output=True, text=True, timeout=60
    )
    if result.returncode != 0:
        print(f"FAIL [{result.returncode}]: {file_path}")
        return None
    try:
        return json.loads(result.stdout)
    except json.JSONDecodeError:
        print(f"INVALID JSON: {file_path}")
        return None

async def batch_analyze(files: list[str], concurrency: int = 4):
    """Process files concurrently with rate limiting."""
    sem = asyncio.Semaphore(concurrency)
    results = {}

    async def process(f: str):
        async with sem:
            proc = await asyncio.create_subprocess_exec(
                "gemini", "-p", f"Review {f} for bugs", "--output-format", "json",
                stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
            )
            stdout, _ = await proc.communicate()
            if proc.returncode == 0:
                results[f] = json.loads(stdout)

    await asyncio.gather(*[process(f) for f in files])
    return results