بناء وكلاء ذكاء اصطناعي إنتاجيين باستخدام Claude Agent SDK للغة بايثون

mkdir claude-agent-python && cd claude-agent-python
python -m venv .venv
source .venv/bin/activate  # على ويندوز: .venv\Scripts\activate
 
pip install claude-agent-sdk

npm install -g @anthropic-ai/claude-code

export ANTHROPIC_API_KEY="sk-ant-..."

import anyio
from claude_agent_sdk import query, ClaudeAgentOptions
 
async def main():
    options = ClaudeAgentOptions(
        system_prompt="You are a concise senior Python engineer.",
        allowed_tools=["Read", "Glob", "Grep"],
        cwd=".",
    )
 
    async for message in query(
        prompt="List the Python files in this folder and summarize what each does.",
        options=options,
    ):
        print(message)
 
anyio.run(main)

python first_agent.py

from claude_agent_sdk import (
    query, ClaudeAgentOptions,
    AssistantMessage, TextBlock, ResultMessage,
)
 
async def main():
    options = ClaudeAgentOptions(allowed_tools=["Read", "Glob"])
 
    async for message in query(prompt="Summarize README.md", options=options):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    print(block.text)
        elif isinstance(message, ResultMessage):
            print(f"\n--- done in {message.num_turns} turns, "
                  f"cost ${message.total_cost_usd:.4f} ---")

import anyio
from claude_agent_sdk import (
    ClaudeSDKClient, ClaudeAgentOptions,
    AssistantMessage, TextBlock,
)
 
async def main():
    options = ClaudeAgentOptions(
        system_prompt="You are a helpful coding assistant.",
        allowed_tools=["Read", "Write", "Edit", "Bash"],
        permission_mode="acceptEdits",
    )
 
    async with ClaudeSDKClient(options=options) as client:
        # الدور الأول
        await client.query("Create a file hello.py that prints today's date.")
        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(block.text)
 
        # دور المتابعة — يتذكّر الوكيل hello.py
        await client.query("Now add a function that returns the date as ISO 8601.")
        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(block.text)
 
anyio.run(main)

import anyio
from claude_agent_sdk import (
    tool, create_sdk_mcp_server,
    ClaudeSDKClient, ClaudeAgentOptions,
    AssistantMessage, TextBlock,
)
 
# مخزن بيانات وهمي للعرض التوضيحي
ORDERS = {
    "A100": {"customer": "Layla", "weight_kg": 2.5, "country": "TN"},
    "A101": {"customer": "Omar", "weight_kg": 8.0, "country": "SA"},
}
 
@tool("get_order", "Fetch an order by its ID", {"order_id": str})
async def get_order(args):
    order = ORDERS.get(args["order_id"])
    if not order:
        return {"content": [{"type": "text", "text": "Order not found"}]}
    return {"content": [{"type": "text", "text": str(order)}]}
 
@tool("shipping_cost", "Estimate shipping cost in USD",
      {"weight_kg": float, "country": str})
async def shipping_cost(args):
    base = 5.0 if args["country"] == "TN" else 12.0
    cost = base + args["weight_kg"] * 1.5
    return {"content": [{"type": "text", "text": f"${cost:.2f}"}]}
 
async def main():
    server = create_sdk_mcp_server(
        name="fulfillment",
        version="1.0.0",
        tools=[get_order, shipping_cost],
    )
 
    options = ClaudeAgentOptions(
        mcp_servers={"fulfillment": server},
        allowed_tools=[
            "mcp__fulfillment__get_order",
            "mcp__fulfillment__shipping_cost",
        ],
    )
 
    async with ClaudeSDKClient(options=options) as client:
        await client.query("What will it cost to ship order A101?")
        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(block.text)
 
anyio.run(main)

from claude_agent_sdk import (
    ClaudeSDKClient, ClaudeAgentOptions,
    PermissionResultAllow, PermissionResultDeny,
    ToolPermissionContext,
)
 
async def gatekeeper(tool_name: str, tool_input: dict,
                     context: ToolPermissionContext):
    # حظر أوامر الصدفة التدميرية مباشرة
    if tool_name == "Bash":
        command = str(tool_input.get("command", ""))
        for danger in ("rm -rf", "sudo", "mkfs", ":(){"):
            if danger in command:
                return PermissionResultDeny(
                    message=f"Blocked dangerous command: {danger}",
                    interrupt=True,
                )
    # كل ما عدا ذلك مسموح
    return PermissionResultAllow()
 
async def main():
    options = ClaudeAgentOptions(
        allowed_tools=["Read", "Write", "Bash"],
        can_use_tool=gatekeeper,
    )
    async with ClaudeSDKClient(options=options) as client:
        await client.query("Clean up temp files in this directory.")
        async for msg in client.receive_response():
            print(msg)

from claude_agent_sdk import ClaudeAgentOptions, HookMatcher
 
async def audit_pre_tool(input_data, tool_use_id, context):
    print(f"[AUDIT] about to run {input_data['tool_name']} "
          f"(id={tool_use_id})")
    return {}  # قاموس فارغ = السماح والمتابعة بشكل طبيعي
 
async def audit_post_tool(input_data, tool_use_id, context):
    print(f"[AUDIT] finished {input_data['tool_name']}")
    return {}
 
options = ClaudeAgentOptions(
    allowed_tools=["Read", "Bash"],
    hooks={
        "PreToolUse": [HookMatcher(hooks=[audit_pre_tool])],
        "PostToolUse": [HookMatcher(hooks=[audit_post_tool])],
    },
)

import anyio
from claude_agent_sdk import (
    tool, create_sdk_mcp_server,
    ClaudeSDKClient, ClaudeAgentOptions,
    HookMatcher, PermissionResultAllow, PermissionResultDeny,
    AssistantMessage, TextBlock, ResultMessage,
)
 
@tool("count_todos", "Count TODO comments in a source file",
      {"path": str})
async def count_todos(args):
    try:
        with open(args["path"], encoding="utf-8") as fh:
            n = sum(1 for line in fh if "TODO" in line)
        return {"content": [{"type": "text", "text": f"{n} TODOs"}]}
    except FileNotFoundError:
        return {"content": [{"type": "text", "text": "File not found"}]}
 
async def gatekeeper(tool_name, tool_input, context):
    if tool_name == "Bash" and "rm " in str(tool_input.get("command", "")):
        return PermissionResultDeny(message="No deletions allowed")
    return PermissionResultAllow()
 
async def audit(input_data, tool_use_id, context):
    print(f"[hook] {input_data['tool_name']}")
    return {}
 
async def main():
    server = create_sdk_mcp_server(name="repo", tools=[count_todos])
    options = ClaudeAgentOptions(
        system_prompt="You are a meticulous code auditor.",
        mcp_servers={"repo": server},
        allowed_tools=["Read", "Glob", "Grep", "mcp__repo__count_todos"],
        can_use_tool=gatekeeper,
        hooks={"PreToolUse": [HookMatcher(hooks=[audit])]},
        max_turns=10,
    )
 
    async with ClaudeSDKClient(options=options) as client:
        await client.query(
            "Find the Python file with the most TODO comments and report it."
        )
        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(block.text)
            elif isinstance(msg, ResultMessage):
                print(f"\nUsage: {msg.usage}")
 
anyio.run(main)