From goal to outcome /
it takes a city.

from ddgs import DDGS
import harnessx as harness

tools = harness.ToolRegistry()


# permission is optional; tools are allowed by default. ASK pauses for an
# approval callback, and DENY blocks the tool.
@tools.register(permission=harness.PermissionLevel.ALLOW)
def search_web(query: str) -> list[dict]:
    """Search the web for titles, URLs, and short excerpts."""
    return DDGS(timeout=10).text(query, max_results=5)


# Change this to whatever you want researched.
TOPIC = "running Python agents in a sandboxed container"

TASK = (
    f"Research {TOPIC} using the search and MCP tools available to you. "
    "Write a Markdown brief covering the steps to adopt it, where the limits "
    "are, and what you could not verify. End with links to the sources you "
    "actually used."
)


async def main():
    # Use Claude, and switch to an OpenAI model if Claude fails temporarily.
    config = harness.AgentConfig(
        provider="anthropic",
        fallbacks=[harness.Fallback("openai", model=os.environ["FALLBACK_MODEL"])],
        retry=harness.RetryPolicy(attempts=3),
    )

    async with harness.MCPManager() as mcp:
        research_server = harness.MCPServerConfig.http(
            "research",
            "https://mcp.example.com/mcp",  # Your MCP server
            permission="allow",  # Allow this server's tools
        )
        await mcp.connect(research_server)

        async with harness.Agent(
            tools=tools,
            mcp=mcp,
            config=config,
        ) as agent:

            # Print the brief as it is written.
            async with agent.run_stream(TASK) as stream:
                async for event in stream:
                    if event.type == harness.RunEventType.TEXT_DELTA:
                        print(event.data, end="", flush=True)
                result = await stream.result()

    result.raise_for_status()
    Path("research.md").write_text(result.output, encoding="utf-8")


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