Pydantic AI

pip install "pydantic-ai-harness[e2b]>=0.54.0" "pydantic-ai-slim[openai]>=2.54.0"

export E2B_API_KEY="your_e2b_api_key"
export OPENAI_API_KEY="your_openai_api_key"

import asyncio

from pydantic_ai import Agent
from pydantic_ai.usage import UsageLimits
from pydantic_ai.workspaces import Workspace
from pydantic_ai_harness.e2b_sandbox import E2BSandbox, E2BSandboxBackend
from pydantic_ai_harness.filesystem import FileSystem
from pydantic_ai_harness.shell import Shell


async def main():
    backend = E2BSandboxBackend(
        working_dir="/home/user/log-cli",
        env={"PYTHONDONTWRITEBYTECODE": "1"},
        sandbox_timeout=300,
        allow_internet_access=False,
    )
    try:
        workspace = Workspace(backend)
        await workspace.write_text("log_stats.py", """import json
from pathlib import Path

records = [json.loads(line) for line in Path('requests.jsonl').read_text().splitlines()]
print(sum(record['status'] == 500 for record in records))
""")
        await workspace.write_text("requests.jsonl", '{"status": 200}\n{"status": 500}\n{"status": 503}\n')
        agent = Agent(
            "openai:gpt-4.1-mini",
            capabilities=[
                E2BSandbox(),
                Shell(tools=["run_command"], default_timeout=20),
                FileSystem(tools=["read_file", "write_file", "edit_file"], content_hashes=False),
            ],
            model_settings={"max_tokens": 1024, "timeout": 30, "parallel_tool_calls": False},
        )
        async with asyncio.timeout(120):
            result = await agent.run(
                "Fix log_stats.py to count all HTTP 500–599 responses as server errors. "
                "Read both files and run the CLI. Do not change requests.jsonl.",
                workspace=backend,
                usage_limits=UsageLimits(request_limit=6, input_tokens_limit=30_000),
            )
        check = await workspace.run(["python", "log_stats.py"], timeout=20)
        assert check.exit_code == 0 and check.stdout.strip() == "2", check
        assert await workspace.read_text("requests.jsonl") == '{"status": 200}\n{"status": 500}\n{"status": 503}\n'
        print(f"Sandbox: {result.workspace.ref.id}; verified server errors: 2")
    finally:
        if backend.ref is not None:
            await E2BSandbox().destroy(backend.ref)


if __name__ == "__main__":
    asyncio.run(main())