import asyncio
from anycode import AnyCode

async def main():
    engine = AnyCode()

    result = await engine.run_agent(
        config={
            "name": "engineer",
            "model": "claude-sonnet-4-6",
            "tools": ["bash", "file_write"],
        },
        prompt="Create a Python utility that checks whether a given string is a palindrome, save it to /tmp/palindrome.py, and execute it.",
    )

    print(result.output)

asyncio.run(main())

import asyncio
from anycode import AnyCode, AgentConfig, TeamConfig

planner = AgentConfig(
    name="planner",
    model="claude-sonnet-4-6",
    system_prompt="You draft module interfaces, folder layouts, and endpoint schemas.",
    tools=["file_write"],
)

builder = AgentConfig(
    name="builder",
    model="claude-sonnet-4-6",
    system_prompt="You translate specifications into production-ready code.",
    tools=["bash", "file_read", "file_write", "file_edit"],
)

auditor = AgentConfig(
    name="auditor",
    model="claude-sonnet-4-6",
    system_prompt="You inspect code for bugs, edge cases, and readability concerns.",
    tools=["file_read", "grep"],
)

async def main():
    engine = AnyCode(config={
        "default_model": "claude-sonnet-4-6",
        "on_progress": lambda ev: print(ev.type, ev.agent or ev.task or ""),
    })

    team = engine.create_team("backend-crew", TeamConfig(
        name="backend-crew",
        agents=[planner, builder, auditor],
        shared_memory=True,
    ))

    result = await engine.run_team(team, "Scaffold a CRUD API for a notes app in /tmp/notes-api/")

    print(f"Completed: {result.success}")
    print(f"Tokens used: {result.total_token_usage.output_tokens}")

asyncio.run(main())

from anycode import TaskSpec

result = await engine.run_tasks(team, [
    TaskSpec(
        title="Draft schema definitions",
        description="Produce Python type declarations and save them to /tmp/types.md",
        assignee="planner",
    ),
    TaskSpec(
        title="Implement core logic",
        description="Read /tmp/types.md and build the service layer in /tmp/lib/",
        assignee="builder",
        depends_on=["Draft schema definitions"],
    ),
    TaskSpec(
        title="Write unit tests",
        description="Author pytest test suites covering all service methods.",
        assignee="builder",
        depends_on=["Implement core logic"],
    ),
    TaskSpec(
        title="Audit implementation",
        description="Examine /tmp/lib/ and generate a detailed review report.",
        assignee="auditor",
        depends_on=["Implement core logic"],
    ),
])

from pydantic import BaseModel, Field
from anycode import define_tool, Agent, ToolRegistry, ToolExecutor, register_built_in_tools, ToolResult, ToolUseContext

class ArticleSearchInput(BaseModel):
    topic: str = Field(description="Subject to search for.")
    limit: int = Field(default=5, description="Maximum articles to return.")

async def fetch_articles(params: ArticleSearchInput, ctx: ToolUseContext) -> ToolResult:
    articles = await my_knowledge_base(params.topic, params.limit)
    return ToolResult(data=json.dumps(articles), is_error=False)

fetch_articles_tool = define_tool(
    name="fetch_articles",
    description="Retrieves relevant articles from the knowledge base.",
    input_model=ArticleSearchInput,
    execute=fetch_articles,
)

registry = ToolRegistry()
register_built_in_tools(registry)
registry.register(fetch_articles_tool)

executor = ToolExecutor(registry)
agent = Agent(
    config={"name": "analyst", "model": "claude-sonnet-4-6", "tools": ["fetch_articles"]},
    tool_registry=registry,
    tool_executor=executor,
)

result = await agent.run("Summarize the latest changes in the Python typing module.")

thinker = AgentConfig(
    name="thinker",
    model="claude-opus-4-6",
    provider="anthropic",
    system_prompt="You devise architectural blueprints and technical strategies.",
    tools=["file_write"],
)

implementer = AgentConfig(
    name="implementer",
    model="gpt-4o",
    provider="openai",
    system_prompt="You transform plans into functional, tested code.",
    tools=["bash", "file_read", "file_write"],
)

team = engine.create_team("cross-provider", TeamConfig(
    name="cross-provider",
    agents=[thinker, implementer],
    shared_memory=True,
))

await engine.run_team(team, "Create a CLI utility that transforms YAML files into JSON format.")

import asyncio
import sys
from anycode import Agent, ToolRegistry, ToolExecutor, register_built_in_tools

async def main():
    registry = ToolRegistry()
    register_built_in_tools(registry)
    executor = ToolExecutor(registry)

    narrator = Agent(
        config={"name": "narrator", "model": "claude-sonnet-4-6", "max_turns": 3},
        tool_registry=registry,
        tool_executor=executor,
    )

    async for ev in narrator.stream("Describe the observer pattern in three sentences."):
        if ev.type == "text" and isinstance(ev.data, str):
            sys.stdout.write(ev.data)

asyncio.run(main())