agent-cow

Database Copy-On-Write for AI agent workspace isolation

agent-cow intercepts your AI agent's database writes and isolates them in a copy-on-write layer. The agent thinks it's modifying real data, but nothing touches production until you approve. Zero changes to your existing queries.

Read the full article: Copy-on-Write in Agentic Systems Try the interactive demo: www.agent-cow.com

Without agent-cow:                With agent-cow:

┌───────┐       ┌──────────┐     ┌───────┐     ┌──────┐     ┌──────────┐
│ Agent │──────>│ Database │     │ Agent │────>│ COW  │────>│ Database │
└───────┘       └──────────┘     └───────┘     │ View │     └──────────┘
 writes directly                               └──────┘
 to production                                   writes go to changes table
                                                 reads merge base + changes
                                                 user reviews, then commits or discards

Installation

pip install agent-cow

Requires Python 3.10+.

How It Works

1. Renames your table from users to users_base
2. Creates a changes table users_changes to store session-specific modifications
3. Creates a COW view named users that merges base + changes
4. Your code doesn't change — queries still target users (now a view)

When you set app.session_id and app.operation_id variables, all writes go to the changes table. Reads automatically merge base data with your session's changes. Other sessions (and production) see only the base data.

See the interactive demo for a worked example of an inventory management system where an agent makes both good and bad decisions.

Backends

Quick Example (PostgreSQL)

import uuid
from agentcow.postgres import deploy_cow_functions, enable_cow_schema, apply_cow_variables, commit_cow_session

# Wrap any async PostgreSQL driver — asyncpg, SQLAlchemy, psycopg, etc.
class MyExecutor:
    def __init__(self, conn):
        self._conn = conn
    async def execute(self, sql: str) -> list[tuple]:
        return [tuple(r) for r in await self._conn.fetch(sql)]

executor = MyExecutor(conn)

# One-time setup — enables COW on all tables in the schema
await deploy_cow_functions(executor)
await enable_cow_schema(executor)

# Agent session — all writes are isolated
session_id = uuid.uuid4()
await apply_cow_variables(executor, session_id, operation_id=uuid.uuid4())
await executor.execute("INSERT INTO users (name) VALUES ('Bessie')")

# Review, then commit or discard
await commit_cow_session(executor, "users", session_id)

See the PostgreSQL docs for the full guide: driver adapters, schema-wide setup, selective commit, web framework integration, and the complete API reference.

API Reference

Core Functions

• deploy_cow_functions(executor) — Deploy COW SQL functions (one-time setup)
• enable_cow(executor, table_name) — Enable COW on a table
• enable_cow_schema(executor) — Enable COW on all tables in a schema
• disable_cow(executor, table_name) — Disable COW and restore original table
• disable_cow_schema(executor) — Disable COW on all tables in a schema
• commit_cow_session(executor, table_name, session_id) — Commit all session changes
• discard_cow_session(executor, table_name, session_id) — Discard all session changes
• get_cow_status(executor) — Get COW status for a schema

Operation-Level Functions

• apply_cow_variables(executor, session_id, operation_id) — Set COW session variables
• get_session_operations(executor, session_id) — List all operations in a session
• get_operation_dependencies(executor, session_id) — Get operation dependency graph
• commit_cow_operations(executor, table_name, session_id, operation_ids) — Commit specific operations
• discard_cow_operations(executor, table_name, session_id, operation_ids) — Discard specific operations

Session Management

• CowPostgresConfig — Dataclass for COW configuration
• build_cow_variable_statements(session_id, operation_id) — Build SET LOCAL SQL statements

For parsing COW configuration from HTTP request headers (e.g. in FastAPI/Django/Flask middleware), see agentcow/examples/header_parsing_example.py.

Development

git clone https://github.com/trail-ml/agent-cow-python.git
cd agent-cow-python
pip install -e ".[dev]"
pytest agentcow/postgres/tests/ -v

Contributing

We welcome contributions! For questions, bug reports, or feature requests, please open an issue.

License

MIT License.

Credits

Created and maintained by trail.