Claude Agent SDK

import asyncio
import os
from claude_agent_sdk import query, ClaudeAgentOptions
from dotenv import load_dotenv
load_dotenv()

ZENROWS_API_KEY = os.environ["ZENROWS_API_KEY"]

async def main() -> None:
    options = ClaudeAgentOptions(
        model="claude-opus-4-7",
        system_prompt=(
            "You are a thorough web researcher. When given a topic and a starting URL, "
            "visit the URL, identify the 3 most relevant linked articles, scrape each one, "
            "and produce a synthesized report covering key findings, common themes, and "
            "any contradictions. Cite each source with its URL."
        ),
        mcp_servers={
            "zenrows": {
                "type": "http",
                "url": "https://mcp.zenrows.com/mcp",
                "headers": {
                    "Authorization": f"Bearer {ZENROWS_API_KEY}",
                },
            }
        },
        allowed_tools=["mcp__zenrows__*"],
        permission_mode="bypassPermissions",
        max_turns=20,
    )

    async for message in query(
        prompt=(
            "Research recent developments in open-source LLMs. "
            "Start at https://huggingface.co/blog and follow the most relevant links."
        ),
        options=options,
    ):
        if hasattr(message, "result"):
            print(message.result)

asyncio.run(main())

import asyncio
import os
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions
from dotenv import load_dotenv
load_dotenv()

ZENROWS_API_KEY = os.environ["ZENROWS_API_KEY"]

async def main() -> None:
    options = ClaudeAgentOptions(
        model="claude-sonnet-4-6",
        system_prompt="You are a web research assistant with access to ZenRows scraping tools.",
        mcp_servers={
            "zenrows": {
                "type": "http",
                "url": "https://mcp.zenrows.com/mcp",
                "headers": {
                    "Authorization": f"Bearer {ZENROWS_API_KEY}",
                },
            }
        },
        allowed_tools=["mcp__zenrows__*"],
        permission_mode="bypassPermissions",
    )

    async with ClaudeSDKClient(options=options) as client:
        # First turn
        await client.query("Visit https://huggingface.co/blog and summarize the top post.")
        async for message in client.receive_response():
            if hasattr(message, "result"):
                print(message.result)

        # Follow-up turn, sharing conversation history
        await client.query("Now scrape the linked GitHub repo and tell me what license it uses.")
        async for message in client.receive_response():
            if hasattr(message, "result"):
                print(message.result)

asyncio.run(main())

import asyncio
import os
from claude_agent_sdk import (
    query,
    tool,
    create_sdk_mcp_server,
    ClaudeAgentOptions,
)
from dotenv import load_dotenv
load_dotenv()

ZENROWS_API_KEY = os.environ["ZENROWS_API_KEY"]

# Custom in-process tool
@tool("save_summary", "Save a research summary to a file", {"filename": str, "content": str})
async def save_summary(args):
    with open(args["filename"], "w") as f:
        f.write(args["content"])
    return {
        "content": [
            {"type": "text", "text": f"Saved to {args['filename']}"}
        ]
    }

local_server = create_sdk_mcp_server(
    name="local",
    version="1.0.0",
    tools=[save_summary],
)

async def main() -> None:
    options = ClaudeAgentOptions(
        model="claude-sonnet-4-6",
        system_prompt=(
            "Research the user's topic with ZenRows, then save your final summary "
            "using the save_summary tool."
        ),
        mcp_servers={
            "zenrows": {
                "type": "http",
                "url": "https://mcp.zenrows.com/mcp",
                "headers": {
                    "Authorization": f"Bearer {ZENROWS_API_KEY}",
                },
            },
            "local": local_server,
        },
        allowed_tools=[
            "mcp__zenrows__*",
            "mcp__local__save_summary",
        ],
        permission_mode="bypassPermissions",
    )

    async for message in query(
        prompt=(
            "Scrape https://huggingface.co/blog, summarize the top three posts, "
            "and save the summary to hf_summary.txt."
        ),
        options=options,
    ):
        if hasattr(message, "result"):
            print(message.result)

asyncio.run(main())