AI
Agents

mkdir tool-agent && cd tool-agent
python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate
pip install anthropic requests python-dotenv

ANTHROPIC_API_KEY=your_anthropic_key_here
SERPER_API_KEY=your_serper_key_here

import os
import json
import requests
from datetime import date
from dotenv import load_dotenv
import anthropic

load_dotenv()

client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))

# --- Tool implementations ---

def search_web(query: str) -> str:
    url = "https://google.serper.dev/search"
    headers = {"X-API-KEY": os.getenv("SERPER_API_KEY"), "Content-Type": "application/json"}
    payload = {"q": query, "num": 5}
    response = requests.post(url, json=payload, headers=headers)
    data = response.json()
    results = []
    for item in data.get("organic", []):
        results.append(f"{item['title']}\n{item['link']}\n{item.get('snippet', '')}")
    return "\n\n".join(results) if results else "No results found."

def write_file(filename: str, content: str) -> str:
    with open(filename, "w") as f:
        f.write(content)
    return f"File '{filename}' written successfully ({len(content)} characters)."

def get_date() -> str:
    return f"Today's date is {date.today().strftime('%B %d, %Y')}."

# --- Tool schemas for Claude ---

TOOLS = [
    {
        "name": "search_web",
        "description": "Search the web for current information on any topic. Returns titles, URLs, and snippets.",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "The search query to look up."}
            },
            "required": ["query"]
        }
    },
    {
        "name": "write_file",
        "description": "Write text content to a file on disk.",
        "input_schema": {
            "type": "object",
            "properties": {
                "filename": {"type": "string", "description": "The name of the file to write, e.g. 'report.txt'."},
                "content": {"type": "string", "description": "The text content to write into the file."}
            },
            "required": ["filename", "content"]
        }
    },
    {
        "name": "get_date",
        "description": "Returns today's current date.",
        "input_schema": {
            "type": "object",
            "properties": {},
            "required": []
        }
    }
]

def run_tool(name: str, tool_input: dict) -> str:
    if name == "search_web":
        return search_web(tool_input["query"])
    elif name == "write_file":
        return write_file(tool_input["filename"], tool_input["content"])
    elif name == "get_date":
        return get_date()
    return "Unknown tool."

def run_agent(task: str):
    print(f"\nTask: {task}\n{'='*50}")
    messages = [{"role": "user", "content": task}]

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

        # Append Claude's response to message history
        messages.append({"role": "assistant", "content": response.content})

        if response.stop_reason == "end_turn":
            # Extract final text response
            for block in response.content:
                if hasattr(block, "text"):
                    print(f"\nAgent: {block.text}")
            break

        # Process tool calls
        tool_results = []
        for block in response.content:
            if block.type == "tool_use":
                print(f"  [Tool] {block.name}({json.dumps(block.input)})")
                result = run_tool(block.name, block.input)
                print(f"  [Result] {result[:200]}{'...' if len(result) > 200 else ''}\n")
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": result
                })

        # Append tool results and loop again
        messages.append({"role": "user", "content": tool_results})

if __name__ == "__main__":
    run_agent(
        "Search for the latest Claude API updates, summarize the key points, "
        "and save the summary to report.txt. Include today's date at the top."
    )

python agent.py

pip install anthropic python-dotenv

import os
from dotenv import load_dotenv
import anthropic

load_dotenv()
client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))

prompt = "A startup has $500k runway, 6 months to profitability, but a competitor just raised $10M. Should they pivot, double down, or seek funding? Analyze carefully."

response = client.messages.create(
    model="claude-opus-4-6",
    max_tokens=16000,
    thinking={"type": "enabled", "budget_tokens": 10000},
    messages=[{"role": "user", "content": prompt}]
)

for block in response.content:
    if block.type == "thinking":
        print("=== Claude's Reasoning ===")
        print(block.thinking)
        print()
    elif block.type == "text":
        print("=== Final Answer ===")
        print(block.text)

python thinking_demo.py

import os
import sys
from dotenv import load_dotenv
import anthropic

load_dotenv()
client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))

# Budget presets — tune based on complexity
DEPTH_BUDGETS = {
    "quick":  4000,   # Simple questions, short docs
    "normal": 10000,  # Most business analysis tasks
    "deep":   20000,  # Complex contracts, multi-factor decisions
}

SYSTEM_PROMPT = """You are a senior business analyst and strategic advisor.
When analyzing documents or problems, structure your response with:
1. Executive Summary (2-3 sentences)
2. Key Findings (bullet points)
3. Risks and Concerns
4. Recommendations (prioritized)
5. Open Questions

Be direct, specific, and actionable."""

def analyze(file_path: str, depth: str = "normal", question: str = None):
    if not os.path.exists(file_path):
        print(f"Error: File '{file_path}' not found.")
        sys.exit(1)

    with open(file_path, "r") as f:
        document = f.read()

    budget = DEPTH_BUDGETS.get(depth, DEPTH_BUDGETS["normal"])

    user_message = f"Document to analyze:\n\n{document}"
    if question:
        user_message += f"\n\nSpecific question: {question}"
    else:
        user_message += "\n\nProvide a comprehensive analysis of this document."

    print(f"Analyzing '{file_path}' at depth='{depth}' (budget_tokens={budget})...")
    print("Thinking...\n")

    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=16000,
        thinking={"type": "enabled", "budget_tokens": budget},
        system=SYSTEM_PROMPT,
        messages=[{"role": "user", "content": user_message}]
    )

    thinking_shown = False
    for block in response.content:
        if block.type == "thinking":
            if not thinking_shown:
                print("=== Internal Reasoning (thinking blocks) ===")
                thinking_shown = True
            # Show a preview — full thinking can be very long
            preview = block.thinking[:1000]
            print(preview + ("..." if len(block.thinking) > 1000 else ""))
            print()
        elif block.type == "text":
            print("=== Analysis ===")
            print(block.text)

    # Save output
    out_path = file_path.replace(".txt", "") + "_analysis.txt"
    final_text = next((b.text for b in response.content if b.type == "text"), "")
    with open(out_path, "w") as f:
        f.write(final_text)
    print(f"\nAnalysis saved to: {out_path}")

if __name__ == "__main__":
    # Usage: python analyze.py  [depth] [question]
    if len(sys.argv) < 2:
        print("Usage: python analyze.py  [quick|normal|deep] [optional question]")
        sys.exit(1)

    file_path = sys.argv[1]
    depth = sys.argv[2] if len(sys.argv) > 2 else "normal"
    question = sys.argv[3] if len(sys.argv) > 3 else None

    analyze(file_path, depth, question)

# Create a quick test document
cat > business_plan.txt << 'EOF'
Company: NovaMesh
Product: AI-powered mesh WiFi that auto-optimizes for each household's usage patterns.
Stage: Pre-seed. $180k raised from angels.
Team: 2 engineers, 1 designer. No sales hire yet.
Traction: 120 beta users, 4.6/5 rating, 0% churn in 3 months.
Market: $12B home networking market growing 8% YoY.
Ask: $1.2M seed to hire sales lead, scale manufacturing, launch in 3 cities.
Competition: Eero (Amazon), Google Nest WiFi, Orbi.
Concern: Manufacturing lead times are 16 weeks, limiting rapid iteration.
EOF

# Quick analysis
python analyze.py business_plan.txt quick

# Deep analysis with a specific question
python analyze.py business_plan.txt deep "Is the $1.2M ask reasonable given their traction?"

import os
import json
import re
from dotenv import load_dotenv
import anthropic

load_dotenv()
client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
MODEL = "claude-sonnet-4-6"

def call_claude(system: str, user: str) -> str:
    response = client.messages.create(
        model=MODEL,
        max_tokens=4096,
        system=system,
        messages=[{"role": "user", "content": user}]
    )
    return response.content[0].text

def extract_json(text: str) -> dict:
    """Strip markdown fences and parse JSON from Claude's response."""
    cleaned = re.sub(r"```(?:json)?", "", text).strip().strip("`").strip()
    return json.loads(cleaned)

# --- Agent 1: Researcher ---

def research_agent(topic: str) -> dict:
    system = """You are a research specialist. Your ONLY output is valid JSON — no explanation, no markdown prose.
Return this exact structure:
{
  "key_points": ["point 1", "point 2", "point 3", "point 4", "point 5"],
  "context": "2-3 sentences of background context",
  "angle": "The most compelling angle or hook for this topic"
}"""
    user = f"Research this topic thoroughly: {topic}"
    raw = call_claude(system, user)
    return extract_json(raw)

# --- Agent 2: Writer ---

def writer_agent(topic: str, research: dict, feedback: str = None) -> str:
    system = """You are an expert content writer. Write clear, engaging articles for a general audience.
Structure: compelling headline, intro hook, 3-4 substantive paragraphs, strong conclusion.
Return ONLY the article text — no JSON, no commentary."""
    user = f"""Topic: {topic}

Research to use:
Key points: {json.dumps(research['key_points'], indent=2)}
Context: {research['context']}
Angle: {research['angle']}"""
    if feedback:
        user += f"\n\nPrevious version was rejected. Reviewer feedback to address:\n{feedback}"
    return call_claude(system, user)

# --- Agent 3: Reviewer ---

def reviewer_agent(topic: str, article: str) -> dict:
    system = """You are a senior editorial reviewer. Evaluate articles rigorously. Your ONLY output is valid JSON.
Return this exact structure:
{
  "score": 7,
  "issues": ["issue 1", "issue 2"],
  "verdict": "pass",
  "improvements": "Specific, actionable instructions for the writer to improve the article."
}
score: integer 1-10. verdict: "pass" if score >= 7, "fail" if below 7."""
    user = f"Topic: {topic}\n\nArticle to review:\n\n{article}"
    raw = call_claude(system, user)
    return extract_json(raw)

# Quick smoke test — add to the bottom of pipeline.py temporarily
if __name__ == "__main__":
    topic = "How large language models are changing software development"
    print("Testing research agent...")
    research = research_agent(topic)
    print(json.dumps(research, indent=2))

    print("\nTesting writer agent...")
    article = writer_agent(topic, research)
    print(article[:500])

    print("\nTesting reviewer agent...")
    review = reviewer_agent(topic, article)
    print(json.dumps(review, indent=2))

def run_pipeline(topic: str, max_retries: int = 3):
    print(f"\n{'='*60}")
    print(f"Pipeline starting: {topic}")
    print(f"{'='*60}\n")

    # Stage 1: Research
    print("[1/3] Research Agent working...")
    research = research_agent(topic)
    print(f"      Found {len(research['key_points'])} key points. Angle: {research['angle'][:80]}...")

    # Stage 2 & 3: Write → Review → Retry loop
    article = None
    feedback = None

    for attempt in range(1, max_retries + 1):
        print(f"\n[2/3] Writer Agent — attempt {attempt}/{max_retries}...")
        article = writer_agent(topic, research, feedback)
        print(f"      Article written ({len(article)} characters).")

        print(f"[3/3] Reviewer Agent scoring...")
        review = reviewer_agent(topic, article)
        score = review["score"]
        verdict = review["verdict"]

        print(f"      Score: {score}/10 — Verdict: {verdict.upper()}")

        if verdict == "pass":
            print(f"\n✓ Pipeline complete on attempt {attempt}!")
            break
        else:
            if attempt < max_retries:
                print(f"      Issues: {', '.join(review['issues'])}")
                print(f"      Sending feedback to writer for revision...")
                feedback = review["improvements"]
            else:
                print(f"\n  Max retries reached. Saving best version (score: {score}/10).")

    # Save final article
    safe_name = topic[:40].replace(" ", "_").replace("/", "-")
    out_file = f"{safe_name}_article.txt"
    with open(out_file, "w") as f:
        f.write(f"Topic: {topic}\n")
        f.write(f"Final score: {review['score']}/10\n")
        f.write("="*60 + "\n\n")
        f.write(article)

    print(f"\nFinal article saved to: {out_file}")
    return article, review

if __name__ == "__main__":
    run_pipeline("How large language models are changing software development")

python pipeline.py

claude mcp

{
  "mcpServers": {
    "filesystem": {
      "command": "npx",
      "args": [
        "-y",
        "@modelcontextprotocol/server-filesystem",
        "/path/to/your/projects"
      ]
    },
    "supabase": {
      "command": "npx",
      "args": [
        "-y",
        "@supabase/mcp-server-supabase@latest",
        "--access-token",
        "YOUR_SUPABASE_ACCESS_TOKEN"
      ]
    }
  }
}

# Ask Claude Code these prompts (no code needed — just type them):

"List the files in my projects folder"

"Show me all the tables in my Supabase database"

"Read the contents of [any file in your projects folder]"

pip install mcp requests

import json
import os
import requests
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp import types

app = Server("my-tools")

@app.list_tools()
async def list_tools() -> list[types.Tool]:
    return [
        types.Tool(
            name="get_weather",
            description="Get the current weather for a city.",
            inputSchema={
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "The city name, e.g. 'London'"}
                },
                "required": ["city"]
            }
        ),
        types.Tool(
            name="save_note",
            description="Save a note to a local file.",
            inputSchema={
                "type": "object",
                "properties": {
                    "title": {"type": "string", "description": "The note title (used as filename)."},
                    "content": {"type": "string", "description": "The note content."}
                },
                "required": ["title", "content"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[types.TextContent]:
    if name == "get_weather":
        city = arguments["city"]
        # Using open-meteo (free, no API key required) via geocoding
        geo = requests.get(
            f"https://geocoding-api.open-meteo.com/v1/search?name={city}&count=1"
        ).json()
        if not geo.get("results"):
            return [types.TextContent(type="text", text=f"City '{city}' not found.")]
        loc = geo["results"][0]
        weather = requests.get(
            f"https://api.open-meteo.com/v1/forecast"
            f"?latitude={loc['latitude']}&longitude={loc['longitude']}"
            f"¤t_weather=true"
        ).json()
        cw = weather["current_weather"]
        result = f"Weather in {city}: {cw['temperature']}°C, wind {cw['windspeed']} km/h"
        return [types.TextContent(type="text", text=result)]

    elif name == "save_note":
        title = arguments["title"]
        content = arguments["content"]
        notes_dir = os.path.expanduser("~/mcp-notes")
        os.makedirs(notes_dir, exist_ok=True)
        safe_title = title.replace(" ", "_").replace("/", "-")
        file_path = os.path.join(notes_dir, f"{safe_title}.txt")
        with open(file_path, "w") as f:
            f.write(f"# {title}\n\n{content}")
        return [types.TextContent(type="text", text=f"Note saved to {file_path}")]

    return [types.TextContent(type="text", text="Unknown tool.")]

if __name__ == "__main__":
    import asyncio
    asyncio.run(stdio_server(app))

{
  "mcpServers": {
    "filesystem": { "...": "..." },
    "supabase": { "...": "..." },
    "my-tools": {
      "command": "python",
      "args": ["/absolute/path/to/server.py"]
    }
  }
}

# Ask Claude Code:

"What's the weather in Tokyo right now?"

"Save a note titled 'Project Ideas' with the content: Build a Claude agent that monitors my inbox"

# Then verify:
ls ~/mcp-notes/

pip install anthropic playwright python-dotenv
python -m playwright install chromium

import os
import base64
import json
from dotenv import load_dotenv
import anthropic
from playwright.sync_api import sync_playwright

load_dotenv()
client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))

def screenshot_to_base64(page) -> str:
    """Take a screenshot of the current page and return as base64."""
    img_bytes = page.screenshot(full_page=False)
    return base64.standard_b64encode(img_bytes).decode("utf-8")

def run_computer_use_agent(url: str, task: str, max_steps: int = 10):
    print(f"\nTask: {task}")
    print(f"URL:  {url}\n")

    with sync_playwright() as p:
        browser = p.chromium.launch(headless=True)
        page = browser.new_page(viewport={"width": 1280, "height": 800})
        page.goto(url, wait_until="networkidle", timeout=30000)
        print("Page loaded. Starting agent loop...")

        messages = []
        tools = [
            {
                "type": "computer_20250124",
                "name": "computer",
                "display_width_px": 1280,
                "display_height_px": 800,
            }
        ]

        # Initial screenshot
        screenshot_b64 = screenshot_to_base64(page)
        messages.append({
            "role": "user",
            "content": [
                {
                    "type": "image",
                    "source": {"type": "base64", "media_type": "image/png", "data": screenshot_b64}
                },
                {"type": "text", "text": f"Task: {task}\n\nHere is the current state of the browser. Complete the task. When done, use the text tool to return your final answer."}
            ]
        })

        for step in range(max_steps):
            print(f"Step {step + 1}/{max_steps}...")
            response = client.messages.create(
                model="claude-sonnet-4-6",
                max_tokens=4096,
                tools=tools,
                messages=messages,
                betas=["computer-use-2025-01-24"],
            )

            messages.append({"role": "assistant", "content": response.content})

            if response.stop_reason == "end_turn":
                for block in response.content:
                    if hasattr(block, "text") and block.text:
                        print(f"\nResult:\n{block.text}")
                        browser.close()
                        return block.text
                break

            # Execute computer use actions
            tool_results = []
            for block in response.content:
                if block.type == "tool_use" and block.name == "computer":
                    action = block.input.get("action")
                    print(f"  Action: {action} {json.dumps({k:v for k,v in block.input.items() if k != 'action'})}")

                    if action == "screenshot":
                        img = screenshot_to_base64(page)
                        tool_results.append({
                            "type": "tool_result",
                            "tool_use_id": block.id,
                            "content": [{"type": "image", "source": {"type": "base64", "media_type": "image/png", "data": img}}]
                        })
                    elif action == "left_click":
                        page.mouse.click(block.input["coordinate"][0], block.input["coordinate"][1])
                        page.wait_for_timeout(500)
                        img = screenshot_to_base64(page)
                        tool_results.append({
                            "type": "tool_result",
                            "tool_use_id": block.id,
                            "content": [{"type": "image", "source": {"type": "base64", "media_type": "image/png", "data": img}}]
                        })
                    elif action == "type":
                        page.keyboard.type(block.input["text"])
                        tool_results.append({"type": "tool_result", "tool_use_id": block.id, "content": "Typed successfully."})
                    elif action == "key":
                        page.keyboard.press(block.input["key"])
                        tool_results.append({"type": "tool_result", "tool_use_id": block.id, "content": "Key pressed."})
                    elif action == "scroll":
                        page.mouse.wheel(0, block.input.get("delta_y", 300))
                        tool_results.append({"type": "tool_result", "tool_use_id": block.id, "content": "Scrolled."})

            if tool_results:
                messages.append({"role": "user", "content": tool_results})

        browser.close()
        return "Max steps reached."

if __name__ == "__main__":
    result = run_computer_use_agent(
        url="https://news.ycombinator.com",
        task="Find the top 3 story titles on the front page and return them as a numbered list."
    )
    print(f"\nFinal answer: {result}")

python web_agent.py

anthropic>=0.40.0
requests>=2.31.0
python-dotenv>=1.0.0

FROM python:3.11-slim

WORKDIR /app

# Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy agent code
COPY agent.py .

# Run the agent
CMD ["python", "agent.py"]

# Add to the bottom of agent.py, replacing the existing __main__ block:
if __name__ == "__main__":
    import logging
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s [%(levelname)s] %(message)s"
    )

    task = os.getenv(
        "AGENT_TASK",
        "Search for today's top AI news, summarize 3 stories, and save to daily_brief.txt"
    )
    logging.info(f"Starting agent with task: {task}")
    run_agent(task)
    logging.info("Agent completed successfully.")

# Install Railway CLI
npm install -g @railway/cli

# Login and initialize
railway login
railway init

# Deploy (Railway detects your Dockerfile automatically)
railway up

ANTHROPIC_API_KEY=your_key_here
SERPER_API_KEY=your_key_here
AGENT_TASK=Search for the latest AI research papers, summarize the top 3, save to report.txt

# Example cron expressions for Railway:
0 8 * * *      # Every day at 8am UTC
0 */6 * * *    # Every 6 hours
0 9 * * 1-5   # Weekdays at 9am UTC

# View live logs from CLI:
railway logs --tail