Agentic AI: The Complete Guide to Autonomous AI Agents

from anthropic import Anthropic

client = Anthropic()

tools = [
    {
        "name": "web_search",
        "description": "Search the web for current information",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "Search query"}
            },
            "required": ["query"]
        }
    },
    {
        "name": "run_python",
        "description": "Execute Python code and return output",
        "input_schema": {
            "type": "object",
            "properties": {
                "code": {"type": "string", "description": "Python code to run"}
            },
            "required": ["code"]
        }
    }
]

def run_agent(goal: str):
    messages = [{"role": "user", "content": goal}]

    while True:
        response = client.messages.create(
            model="claude-opus-4-5",
            max_tokens=4096,
            tools=tools,
            messages=messages
        )

        # Check if agent wants to use a tool
        if response.stop_reason == "tool_use":
            tool_call = next(b for b in response.content if b.type == "tool_use")
            tool_result = execute_tool(tool_call.name, tool_call.input)

            # Add assistant response + tool result to conversation
            messages.append({"role": "assistant", "content": response.content})
            messages.append({
                "role": "user",
                "content": [{"type": "tool_result", "tool_use_id": tool_call.id, "content": tool_result}]
            })
        else:
            # Agent is done
            final_text = next(b.text for b in response.content if b.type == "text")
            return final_text

# Orchestrator-worker pattern
class OrchestratorAgent:
    def __init__(self):
        self.workers = {
            "researcher": ResearchAgent(),
            "coder": CodeAgent(),
            "writer": WriterAgent()
        }

    def execute(self, task: str) -> str:
        # Step 1: Plan using LLM
        plan = self.plan_task(task)

        results = {}
        for step in plan.steps:
            # Delegate to appropriate specialist
            worker = self.workers[step.agent_type]
            result = worker.execute(step.instruction, context=results)
            results[step.name] = result

        # Synthesize final output
        return self.synthesize(task, results)

# Parallel execution for independent subtasks
import asyncio

async def parallel_agents(tasks: list[dict]) -> list[str]:
    async def run_agent(agent, task):
        return await agent.execute_async(task)

    return await asyncio.gather(*[
        run_agent(agent_map[t["agent"]], t["task"])
        for t in tasks
    ])

class SafeAgent:
    def __init__(self, max_iterations=20):
        self.max_iterations = max_iterations
        self.HIGH_RISK_TOOLS = {"delete_file", "send_email", "make_purchase"}

    def execute(self, goal: str) -> str:
        messages = [{"role": "user", "content": goal}]
        iterations = 0

        while iterations < self.max_iterations:
            iterations += 1
            response = self.call_llm(messages)

            if response.stop_reason != "tool_use":
                return self.extract_text(response)

            tool_call = self.get_tool_call(response)

            # Gate high-risk tools behind human confirmation
            if tool_call.name in self.HIGH_RISK_TOOLS:
                confirmed = self.request_human_approval(tool_call)
                if not confirmed:
                    return "Action cancelled by user."

            result = self.execute_tool(tool_call)
            messages = self.update_messages(messages, response, tool_call.id, result)

        return "Max iterations reached. Task incomplete."

    def request_human_approval(self, tool_call) -> bool:
        print(f"⚠️  Agent wants to run: {tool_call.name}")
        print(f"Parameters: {tool_call.input}")
        response = input("Allow? (yes/no): ")
        return response.lower() == "yes"