Deep Agent

import asyncio
import os
import shutil
import tempfile

from dotenv import load_dotenv

import synalinks

# =============================================================================
# Workspace setup
# =============================================================================


def populate_workspace(workdir: str) -> None:
    """Seed the workdir with a tiny Python project for the agent to work on."""
    os.makedirs(workdir, exist_ok=True)

    # A starter module the agent will extend.
    with open(os.path.join(workdir, "calculator.py"), "w") as f:
        f.write('def add(a, b):\n    """Return a + b."""\n    return a + b\n')

    # A README the agent can read to understand context.
    with open(os.path.join(workdir, "README.md"), "w") as f:
        f.write(
            "# Calculator\n"
            "\n"
            "A tiny module. Currently supports `add(a, b)`.\n"
            "Planned: `multiply(a, b)`.\n"
        )


def print_messages(result) -> None:
    """Pretty-print the agent's tool-call trajectory."""
    messages = result.get("messages", [])
    for msg in messages:
        role = msg.get("role")
        if role == "assistant" and msg.get("tool_calls"):
            for call in msg["tool_calls"]:
                args = call.get("arguments", {})
                args_str = ", ".join(f"{k}={v!r}" for k, v in args.items())
                print(f"  Tool call: {call['name']}({args_str})")
        elif role == "tool":
            content = msg.get("content", "")
            if isinstance(content, dict):
                content = str(content)
            if len(content) > 200:
                content = content[:200] + "..."
            print(f"  Tool result: {content}")
        elif role == "assistant" and msg.get("content"):
            print(f"  Assistant: {msg['content']}")


# =============================================================================
# Main example
# =============================================================================


async def main():
    """Run a small end-to-end task with the deep agent."""
    load_dotenv()
    synalinks.clear_session()

    workdir = tempfile.mkdtemp(prefix="deep_agent_demo_")
    print(f"Workdir: {workdir}\n")

    try:
        populate_workspace(workdir)

        lm = synalinks.LanguageModel(model="gemini/gemini-3.1-flash-lite-preview")

        # Keep a reference to the DeepAgent so we can inspect its sandbox
        # (the copy-on-write overlay) after the run.
        deep_agent = synalinks.DeepAgent(
            workdir=workdir,
            language_model=lm,
            max_iterations=10,
        )
        inputs = synalinks.Input(data_model=synalinks.ChatMessages)
        outputs = await deep_agent(inputs)

        agent = synalinks.Program(
            inputs=inputs,
            outputs=outputs,
            name="deep_agent",
            description="A coding agent with a sandboxed copy of a workdir.",
        )

        agent.summary()

        # ---------------------------------------------------------------------
        # Task 1: explore the workdir
        # ---------------------------------------------------------------------
        print("\n" + "=" * 60)
        print("Task 1: Explore")
        print("=" * 60)

        question = synalinks.ChatMessages(
            messages=[
                synalinks.ChatMessage(
                    role="user",
                    content=(
                        "What's in this directory? List the files and tell "
                        "me what the project is about. Be brief."
                    ),
                )
            ]
        )
        result = await agent(question)
        print_messages(result)

        # ---------------------------------------------------------------------
        # Task 2: extend the calculator (in the overlay, host-safe)
        # ---------------------------------------------------------------------
        print("\n" + "=" * 60)
        print("Task 2: Extend (overlay edit)")
        print("=" * 60)

        task = synalinks.ChatMessages(
            messages=[
                synalinks.ChatMessage(
                    role="user",
                    content=(
                        "Open calculator.py and add a `multiply(a, b)` function "
                        "that returns a * b, in the same style as `add`. Then "
                        "read the file back to confirm the change."
                    ),
                )
            ]
        )
        result = await agent(task)
        print_messages(result)

        # The edit lives in the sandbox overlay — show it, and show that the
        # file on disk is untouched.
        print("\n--- calculator.py (sandbox overlay) ---")
        overlay = await deep_agent.sandbox.read_file("/calculator.py")
        print(overlay.get("content", overlay))

        print("\n--- calculator.py (on disk, unchanged) ---")
        with open(os.path.join(workdir, "calculator.py")) as f:
            print(f.read())

        print("--- overlay changes ---")
        print(deep_agent.sandbox.changes())

    finally:
        shutil.rmtree(workdir, ignore_errors=True)


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