Long-running research agent with durable replay

import asyncio
from datetime import timedelta
from loomflow import Agent, tool
from loomflow.governance.budget import BudgetConfig, StandardBudget
from loomflow.mcp import MCPRegistry, MCPServerSpec
from loomflow.model.anthropic import AnthropicModel
from loomflow.runtime import SqliteRuntime
 
@tool
async def web_search(query: str) -> str:
    """Search the web."""
    ...
 
@tool
async def fetch(url: str) -> str:
    """Fetch a URL."""
    ...
 
mcp_servers = MCPRegistry([
    MCPServerSpec.stdio("git", "uvx", ["mcp-server-git", "--repo", "."]),
    MCPServerSpec.stdio("fs",  "uvx", ["mcp-server-filesystem", "--root", "."]),
])
 
async def main():
    runtime = SqliteRuntime("./research_journal.db")
    agent = Agent(
        "You are a research assistant. Plan a multi-step research task, "
        "execute each step with the available tools, then summarize.",
        model=AnthropicModel("claude-opus-4-7"),
        runtime=runtime,
        tools=[web_search, fetch, *mcp_servers.list_tools_sync()],
        budget=StandardBudget(BudgetConfig(
            max_tokens=500_000,
            max_cost_usd=20.0,
            max_wall_clock=timedelta(hours=2),
        )),
    )
    result = await agent.run("Research the state of agent harnesses in 2026.")
    print(result.output)
 
asyncio.run(main())