Agent SDK — Python — בניית סוכנים בפייתון

מה זה Agent SDK ולמה זה חשוב

עד עכשיו, בכל פעם שרציתם ש-Claude Code יעשה משהו, הפעלתם אותו מ-Terminal: או אינטראקטיבית (הקלדתם פרומפט וקיבלתם תשובה) או ב-Headless Mode (claude -p "..."). זה עובד מצוין — אבל יש גבול. כשאתם רוצים לבנות מערכת שלמה שמשתמשת ב-Claude כמנוע, אתם צריכים משהו אחר. אתם צריכים להפעיל Claude מתוך הקוד שלכם.

ה-Agent SDK (ערכת פיתוח סוכנים) הוא בדיוק זה: ספריית Python שמספקת את אותם הכלים, אותה לולאת סוכן, ואותו ניהול הקשר שיש ב-Claude Code — אבל כ-import בקובץ Python שלכם. במקום לכתוב claude -p "analyze this file" ב-shell, אתם כותבים:

from claude_agent_sdk import Agent

agent = Agent()
result = agent.run("analyze this file")
print(result.text)

זה נשמע כמו הבדל קטן, אבל ההשלכות עצומות. כשClaude רץ מתוך הקוד שלכם, אתם יכולים:

• לעבד את התוצאות תוכנתית: לנתח JSON, לשמור למסד נתונים, לשלוח מייל, לעדכן דשבורד
• לשלוט ב-flow: להריץ סוכנים בתנאים, בלולאות, בזה אחר זה או במקביל
• להגביל ולאבטח: להגדיר בדיוק אילו כלים זמינים, לאיזה קבצים הסוכן יכול לגשת, ומה ה-timeout
• לבנות מוצרים: CLI tools, שירותי backend, pipelines, מערכות monitoring — כל מה שאפשר לחלום

למה לא פשוט API ישיר?

שאלה טובה. Anthropic מציעים גם Anthropic Python API (pip install anthropic) — שמאפשר לשלוח הודעות ולקבל תשובות. אבל ה-API הישיר נותן לכם העברת הודעות בלבד. אתם צריכים לבנות לבד את כל מה שמסביב: Tool Use, Agent Loop, Context Management, Error Recovery, Retries, File Operations, Shell Commands.

ה-Agent SDK נותן לכם את כל זה מובנה. הסוכן יודע לקרוא קבצים, לערוך אותם, להריץ פקודות, לחפש ב-codebase, לגלוש באינטרנט — בלי שאתם כותבים שורת קוד אחת לטיפול ב-tools. Claude מחליט לבד מתי להשתמש בכלי, מפעיל אותו, רואה את התוצאה, ומחליט מה לעשות הלאה.

ה-Agent SDK הוא הכלי שמאפשר לכם להפוך את Claude Code ממשהו שאתם משתמשים בו למשהו שאתם בונים איתו. זה המעבר ממשתמש למפתח. ובפרק הזה — אנחנו עושים את המעבר הזה צעד אחרי צעד.

Use Cases — מה אפשר לבנות?

כדי שזה לא יישאר תיאורטי, הנה 5 דוגמאות מהעולם האמיתי שאנשים בונים עם Agent SDK:

• כלי CLI מותאם: סטארטאפ ישראלי בתל אביב שבנה כלי שמנתח Pull Requests ומוצא בעיות אבטחה ספציפיות לפרויקט שלהם — מעבר למה ש-claude-code-action עושה, כי הם הוסיפו לוגיקה מותאמת. לפי הערכתם זה חוסך כ-15 שעות עבודה בשבוע (שזה כ-3,000 ש"ח לשבוע בשכר מפתח ישראלי)
• שירות Backend: API endpoint שמקבל טקסט חופשי מלקוח ומחזיר JSON מובנה — Claude מנתח, מסווג, ומחלץ entities. הכל בריצה אחת של agent.run()
• Data Pipeline: סוכן שמעבד קבצי נתונים כל לילה — קורא CSV, מנקה, מוודא איכות, ומייצר דו"ח. נבנה אחד כזה בהמשך הפרק
• Monitoring Agent: סוכן שבודק לוגים ומדדי מערכת, מזהה חריגות, ושולח התראות חכמות שמסבירות למה יש בעיה ומה לעשות — גם אותו נבנה בפרק הזה
• כלי Onboarding: סוכן שמפתח חדש מריץ, שקורא את כל ה-codebase ומייצר מדריך התמצאות מותאם — "הנה הקבצים החשובים, הנה ה-patterns שאנחנו משתמשים בהם, הנה איך להרים סביבה מקומית"

שימו לב: כל אחד מהם היה אפשרי גם עם claude -p ב-shell, אבל ה-SDK מוסיף שליטה תוכנתית: תנאים, לולאות, error handling, שמירת תוצאות, שילוב עם מערכות אחרות. זה ההבדל בין סקריפט shell לבין תוכנה אמיתית.

התקנה והסוכן הראשון

הכל מתחיל ב-pip install. ה-Agent SDK זמין ב-PyPI (ספריית החבילות של Python) ודורש Python 3.10 ומעלה. אם אתם לא בטוחים איזו גרסה יש לכם, בדקו עם python3 --version.

צעד 1: התקנה

# התקנה בסיסית
pip install claude-agent-sdk

# או עם גרסה מוצמדת (מומלץ לפרודקשן)
pip install claude-agent-sdk==0.1.48

# ב-virtual environment (מומלץ תמיד)
python3 -m venv agent-env
source agent-env/bin/activate      # Linux/Mac
# agent-env\Scripts\activate       # Windows
pip install claude-agent-sdk==0.1.48

צעד 2: הגדרת API Key

ה-SDK צריך מפתח API של Anthropic כדי לתקשר עם Claude. יש 3 דרכים להגדיר:

# דרך 1: משתנה סביבה (הכי נפוץ)
export ANTHROPIC_API_KEY="sk-ant-your-key-here"

# דרך 2: קובץ .env (עם python-dotenv)
# בקובץ .env:
# ANTHROPIC_API_KEY=sk-ant-your-key-here

# דרך 3: בקוד (רק לפיתוח מקומי! לעולם לא ב-Git)
agent = Agent(api_key="sk-ant-your-key-here")

צעד 3: הסוכן הראשון

הנה הסוכן הכי פשוט שאפשר לבנות. שמרו אותו כ-my_first_agent.py:

from claude_agent_sdk import Agent

# יצירת סוכן עם הגדרות ברירת מחדל
agent = Agent(
    model="claude-sonnet-4-6",   # מודל — Sonnet לפיתוח, Opus לפרודקשן
    max_turns=10,                 # מקסימום סיבובים (מגן מפני לולאות)
    max_tokens=4096               # מקסימום טוקנים בתשובה
)

# הפעלת הסוכן
result = agent.run("List all Python files in the current directory and tell me what each one does")

# הדפסת התוצאה
print(result.text)
print(f"\nTokens used: {result.tokens_used}")
print(f"Cost: ${result.cost:.4f}")
print(f"Tool calls: {len(result.tool_calls)}")

מה קורה כשמריצים python3 my_first_agent.py?

1. ה-SDK יוצר שיחה חדשה עם Claude
2. Claude מקבל את ה-Prompt ("List all Python files...")
3. Claude מחליט לבד שהוא צריך להשתמש בכלי Glob כדי למצוא קבצי .py
4. ה-SDK מפעיל את Glob, מקבל רשימת קבצים, ומחזיר ל-Claude
5. Claude מחליט שהוא צריך לקרוא כל קובץ עם Read
6. ה-SDK מפעיל Read על כל קובץ ומחזיר את התוכן
7. Claude מנתח, מסכם, ומחזיר תשובה

כל זה קורה אוטומטית. אתם לא כתבתם שורת קוד אחת שאומרת "תשתמש ב-Glob" או "תקרא את הקובץ הזה". Claude מחליט לבד מה הוא צריך — ה-SDK מספק את הכלים וה-Agent Loop דואג לריצה.

Sync מול Async

כברירת מחדל, agent.run() הוא סינכרוני — הוא חוסם את ה-thread עד ש-Claude מסיים. לרוב המקרים זה מספיק. אבל אם אתם בונים שרת, API, או כל דבר שצריך לטפל בכמה דברים במקביל — יש אלטרנטיבה אסינכרונית:

import asyncio
from claude_agent_sdk import Agent

async def main():
    agent = Agent(model="claude-sonnet-4-6")
    result = await agent.arun("analyze this code")  # arun = async run
    print(result.text)

asyncio.run(main())

arun() עובד בדיוק כמו run() — אבל לא חוסם. שימושי כשרוצים להריץ כמה סוכנים במקביל או כשהסוכן רץ בתוך שרת אסינכרוני (FastAPI, aiohttp).

ריצת כמה סוכנים במקביל

יתרון גדול של ה-SDK מול CLI: אפשר להריץ כמה סוכנים במקביל. כל אחד עובד על משימה אחרת, ואתם מחכים שכולם יסיימו:

import asyncio
from claude_agent_sdk import Agent

async def run_parallel_agents():
    agent = Agent(model="claude-sonnet-4-6")

    # 3 משימות — רצות במקביל
    tasks = [
        agent.arun("check code quality of src/auth/"),
        agent.arun("analyze test coverage of src/api/"),
        agent.arun("review documentation in docs/"),
    ]

    results = await asyncio.gather(*tasks)

    for i, result in enumerate(results, 1):
        print(f"\n=== Task {i} ===")
        print(f"Cost: ${result.cost:.4f}")
        print(result.text[:200])

asyncio.run(run_parallel_agents())

שלושת הסוכנים רצים בו-זמנית. במקום 3 דקות (דקה לכל אחד ברצף), זה לוקח דקה אחת. העלות זהה — אבל הזמן חצוי. שימו לב שכל arun() הוא סשן עצמאי — הסוכנים לא חולקים context ביניהם.

מושגי ליבה: Tools, Agent Loop, ו-Context

לפני שבונים סוכנים מורכבים, חשוב להבין 3 מושגים שמניעים את כל המערכת. אם תבינו אותם — כל השאר הוא יישום.

1. Tools — הכלים

לסוכן יש גישה לכלים מובנים — אותם הכלים בדיוק שיש ב-Claude Code האינטראקטיבי:

אתם לא צריכים להגיד ל-Claude איזה כלי להשתמש. אתם נותנים לו משימה, והוא בוחר לבד. אם ביקשתם "find all Python files with errors" — הוא ישתמש ב-Glob למציאה ו-Bash להרצת linter. אם ביקשתם "read the config and fix the port" — הוא ישתמש ב-Read ואז Edit.

2. Agent Loop — לולאת הסוכן

לולאת הסוכן היא הלב הפועם של ה-SDK. זה מה שהופך agent.run("...") מ-"שאל שאלה וקבל תשובה" ל-"תן משימה וקבל תוצאה". הנה הלולאה:

1. קבל Prompt: "analyze the test failures and fix them"
2. נתח: Claude חושב מה צריך לעשות
3. בחר כלי: Claude מחליט — "אני צריך להריץ pytest קודם"
4. הפעל כלי: ה-SDK מריץ pytest ומקבל פלט
5. צפה בתוצאה: Claude רואה את שגיאות הטסטים
6. החלט: "עוד לא סיימתי — אני צריך לקרוא את הקובץ הבעייתי" → חוזר לצעד 3
7. סיום: Claude מחליט שהמשימה הושלמה → מחזיר תשובה סופית

הלולאה נעצרת באחד מ-4 מצבים:

• Claude קובע שהמשימה הושלמה
• הגענו ל-max_turns — מנגנון בטיחות
• שגיאה שאי-אפשר להתאושש ממנה
• ביטול תוכנתי מהקוד שלכם

3. Context — ההקשר

כל הודעה, כל קריאת כלי, כל תוצאה — נשמרת ב-context (הקשר) של השיחה. בדיוק כמו בסשן אינטראקטיבי, הסוכן "זוכר" מה הוא עשה קודם. זה מאפשר שיחות מתמשכות:

agent = Agent(model="claude-sonnet-4-6")

# ריצה ראשונה
result1 = agent.run("read the config.json file")
print(result1.text)

# ריצה שנייה — ממשיכה מאותו context
result2 = agent.run(
    "now change the port to 8080",
    messages=result1.messages  # מעביר את ההיסטוריה
)
print(result2.text)

בריצה השנייה, Claude כבר יודע מה יש ב-config.json — הוא קרא אותו קודם. הוא לא צריך לקרוא שוב. זה חוסך טוקנים ומייצר שיחה טבעית ורצופה.

הגבלת כלים — לסוכן בטוח

לא כל סוכן צריך גישה לכל הכלים. סוכן שתפקידו לנתח קוד לא צריך יכולת כתיבה. סוכן שמריץ בסביבת ייצור לא צריך גישה ל-Bash. ה-SDK נותן שליטה מלאה:

# סוכן קריאה בלבד — לא יכול לשנות שום דבר
readonly_agent = Agent(
    model="claude-sonnet-4-6",
    allowed_tools=["Read", "Grep", "Glob"],
    max_turns=15
)

# סוכן בלי Bash — יכול לקרוא ולכתוב קבצים, אבל לא להריץ פקודות
safe_agent = Agent(
    model="claude-sonnet-4-6",
    disallowed_tools=["Bash"],
    max_turns=10
)

# סוכן עם תיקיית עבודה מוגבלת
sandboxed_agent = Agent(
    model="claude-sonnet-4-6",
    working_directory="/project/src",
    allowed_paths=["/project/src", "/project/tests"]
)

allowed_tools מגדיר רשימה לבנה — רק הכלים האלה זמינים. disallowed_tools מגדיר רשימה שחורה — כל הכלים זמינים חוץ מאלה. allowed_paths מגביל לאיזה תיקיות הסוכן יכול לגשת — כמו סנדבוקס.

עבודה עם תוצאות

agent.run() מחזיר אובייקט תוצאה (result) עשיר — לא רק טקסט. הנה מה שיש בו:

result = agent.run("analyze this project")

# 1. הטקסט הסופי שClaude כתב
print(result.text)

# 2. כל הקריאות לכלים — מה Claude עשה בפועל
for call in result.tool_calls:
    print(f"Tool: {call.tool}, Input: {call.input[:80]}...")

# 3. צריכת טוקנים — מפורטת
print(f"Input tokens:  {result.tokens_used.input}")
print(f"Output tokens: {result.tokens_used.output}")
print(f"Thinking tokens: {result.tokens_used.thinking}")

# 4. עלות — בדולרים
print(f"Cost: ${result.cost:.4f}")

# 5. היסטוריית השיחה — לשימוש בהמשך או ל-debug
print(f"Messages count: {len(result.messages)}")

result.tool_calls — ה-Audit Trail שלכם

בפרודקשן, result.tool_calls הוא קריטי. הוא מראה בדיוק מה הסוכן עשה: אילו קבצים קרא, אילו פקודות הריץ, אילו שינויים ביצע. זה ה-audit trail שלכם — היכולת לחזור אחורה ולהבין מה קרה.

# לוג מפורט של כל פעולה
for i, call in enumerate(result.tool_calls, 1):
    print(f"Step {i}: {call.tool}")
    if call.tool == "Read":
        print(f"  File: {call.input.get('file_path', 'N/A')}")
    elif call.tool == "Bash":
        print(f"  Command: {call.input.get('command', 'N/A')}")
    elif call.tool == "Edit":
        print(f"  File: {call.input.get('file_path', 'N/A')}")
        print(f"  Changed: '{call.input.get('old_string', '')[:50]}...'")

# שמירת audit trail לקובץ
import json
audit = [{"step": i, "tool": c.tool, "input": c.input}
         for i, c in enumerate(result.tool_calls, 1)]
with open("audit_trail.json", "w") as f:
    json.dump(audit, f, indent=2, ensure_ascii=False)

result.messages — שימור שיחה בין ריצות

כל ריצה של agent.run() מייצרת היסטוריית שיחה מלאה. אפשר לשמור אותה, להעביר אותה לריצה הבאה, ואפילו לשמור ל-database כדי "לחדש" שיחה ימים מאוחר יותר:

# שמירת היסטוריית שיחה
import json

result = agent.run("analyze the project structure")
with open("conversation_state.json", "w") as f:
    json.dump(result.messages, f)

# ... שעות/ימים אחר כך ...

# טעינה והמשך
with open("conversation_state.json") as f:
    prev_messages = json.load(f)

result2 = agent.run("based on your analysis, suggest improvements",
                     messages=prev_messages)

זה חזק במיוחד ל-multi-stage pipelines: Stage 1 מנתח, שומר state. Stage 2 טוען state ופועל על סמך הניתוח. בלי לשלם שוב על קריאת כל הקבצים.

Streaming — תוצאות בזמן אמת

אם המשימה ארוכה (Claude מעבד 20 קבצים), אתם לא רוצים לחכות 2 דקות ואז לראות הכל בבת אחת. Streaming מאפשר לראות את הפלט מגיע חתיכה-חתיכה:

# במקום agent.run() — agent.stream()
for chunk in agent.stream("analyze all Python files in src/"):
    print(chunk.text, end="", flush=True)
# הפלט מגיע בזמן אמת, מילה אחרי מילה

Streaming שימושי במיוחד ל-UIs, CLIs עם progress, ו-web APIs שרוצים להחזיר Server-Sent Events.

Error Handling בסיסי

סוכנים יכולים להיכשל. ה-API יכול להחזיר שגיאה, כלי יכול לנפול, Claude יכול להתבלבל. תמיד עטפו ב-try/except:

from claude_agent_sdk import Agent, AgentError

agent = Agent(model="claude-sonnet-4-6", max_turns=10)

try:
    result = agent.run("run the test suite and fix any failures")
    print(result.text)
    print(f"Cost: ${result.cost:.4f}")
except AgentError as e:
    print(f"Agent error: {e}")
    # לוג, התראה, retry — תלוי במצב
except Exception as e:
    print(f"Unexpected error: {e}")

פעולות קבצים ב-SDK

פעולות קבצים ב-SDK עובדות בדיוק כמו ב-Claude Code הרגיל. ההבדל היחיד: אתם שולטים בתהליך מהקוד שלכם. Claude משתמש ב-Read, Edit, Write אוטומטית כשהוא צריך — אתם לא צריכים להגיד לו "use the Read tool".

קריאת קבצים

agent = Agent(model="claude-sonnet-4-6")

# Claude מחליט לבד להשתמש ב-Read
result = agent.run("read config.json and tell me what port the server uses")
print(result.text)
# "The server is configured to use port 3000, as defined in config.json line 5"

יצירת וכתיבת קבצים

result = agent.run("""
    Create a new Python file called utils.py with:
    - A function validate_email that checks email format
    - A function format_phone that formats Israeli phone numbers (05X-XXXXXXX)
    - Type hints and docstrings in Hebrew
""")
print(result.text)
# Claude יוצר את הקובץ עם Write tool

עריכת קבצים — כירורגית

# Claude משתמש ב-Edit tool — old_string/new_string
result = agent.run("in config.json, change the port from 3000 to 8080")
# Claude מזהה את השורה, מחליף רק אותה, משאיר הכל השאר

Working Directory ו-Allowed Paths

# הסוכן "רואה" רק את תיקיית הפרויקט
agent = Agent(
    model="claude-sonnet-4-6",
    working_directory="/home/user/my-project",
    allowed_paths=["/home/user/my-project/src", "/home/user/my-project/tests"]
)

# Claude לא יכול לגשת ל-/etc/passwd או כל מקום מחוץ ל-allowed_paths

ה-working_directory קובע מאיפה פקודות Bash רצות ואיפה נתיבים יחסיים מתחילים. allowed_paths מגביל לאילו תיקיות הסוכן יכול לגשת — סנדבוקס ברמת הקבצים.

Bulk File Operations — עיבוד מרובה

אחד היתרונות הגדולים של ה-SDK: אפשר לתת לסוכן משימה שכוללת הרבה קבצים, והוא מטפל בכולם. לא צריך לולאה ב-Python — Claude עושה את הלולאה לבד בתוך ה-Agent Loop:

result = agent.run("""
    Read all .py files in src/.
    For each file:
    - Add type hints to function parameters and return values
    - Add Hebrew docstrings to functions that don't have them
    - Save the modified file (overwrite)
    Report: how many files modified, how many functions updated.
""")
print(result.text)
# Claude ישתמש ב-Glob למציאה, Read לקריאה, Edit לעדכון — על כל קובץ

שימו לב: אם יש 50 קבצים, ה-max_turns צריך להיות גבוה מספיק (כל קריאה + עריכה = 2 turns לקובץ = 100 turns). כלל אצבע: max_turns = מספר_קבצים x 3 (עם buffer).

פקודות Shell ואינטגרציית מערכת

כלי ה-Bash נותן לסוכן יכולת להריץ כל פקודה שאתם הייתם מריצים ב-Terminal. בפרק 1 עשיתם את זה עם claude -p "run tests" — עכשיו אותו הדבר מתוך קוד Python.

agent = Agent(model="claude-sonnet-4-6")

# Claude מחליט לבד להשתמש ב-Bash כשצריך
result = agent.run("run the test suite and report which tests failed")

# Claude ירוץ: pytest (או npm test, או make test — תלוי בפרויקט)
# יראה את הפלט, ינתח אותו, ויחזיר דו"ח מובנה
print(result.text)

הסוכן לא רק מריץ פקודות — הוא מבין את התוצאה. אם pytest נכשל, Claude קורא את ה-traceback, מזהה את הבעיה, ויכול אפילו לתקן אותה (אם יש לו גישה ל-Edit). אם פקודת curl מחזירה 404, Claude מבין שה-endpoint לא קיים.

Sandbox — ברירת המחדל הבטוחה

כברירת מחדל, הסוכן רץ ב-sandbox mode — בידוד ברמת מערכת הפעלה (Seatbelt ב-macOS, bubblewrap ב-Linux). זה אומר:

• הגבלת גישה לקבצים — רק תיקיית העבודה
• הגבלת רשת — רק domains מותרים
• הגבלות על subprocess — אותן מגבלות חלות על כל פקודה שהסוכן מריץ

# sandbox פועל כברירת מחדל — אין צורך להגדיר
agent = Agent(model="claude-sonnet-4-6")  # sandbox=True

# כיבוי sandbox — רק ב-Docker או סביבה מבוקרת!
docker_agent = Agent(
    model="claude-sonnet-4-6",
    sandbox=False,
    skip_permissions=True  # רק ב-Docker/air-gapped!
)

Timeout ו-Long-Running Commands

# timeout כללי לסוכן — מקסימום 5 דקות
agent = Agent(
    model="claude-sonnet-4-6",
    timeout=300  # שניות
)

# אם המשימה ארוכה — Claude יפרק אותה לצעדים קטנים
result = agent.run("run the full integration test suite (might take a few minutes)")
# אם timeout נגמר — AgentError

MCP Integration — שילוב כלים חיצוניים

MCP (Model Context Protocol) הוא תקן פתוח שמאפשר לסוכנים להתחבר לכלים חיצוניים — מסדי נתונים, APIs, שירותים, דפדפן. ב-Claude Code האינטראקטיבי השתמשתם ב-MCP דרך .claude/settings.json. ב-SDK — מגדירים בקוד:

agent = Agent(
    model="claude-sonnet-4-6",
    mcp_servers={
        "playwright": {
            "command": "npx",
            "args": ["@anthropic-ai/mcp-playwright"]
        },
        "postgres": {
            "command": "npx",
            "args": ["@anthropic-ai/mcp-postgres"],
            "env": {"DATABASE_URL": "postgresql://localhost/mydb"}
        }
    }
)

עכשיו לסוכן יש גישה ל-כלי MCP בנוסף לכלים המובנים. הוא יכול לגלוש באינטרנט עם Playwright, לבצע שאילתות SQL ב-PostgreSQL, או כל דבר אחר שיש לו MCP server.

דוגמה: סוכן שגולש ומעבד

# סוכן שמשלב Playwright (גלישה) עם כלים מובנים (כתיבת קבצים)
agent = Agent(
    model="claude-sonnet-4-6",
    mcp_servers={
        "playwright": {
            "command": "npx",
            "args": ["@anthropic-ai/mcp-playwright"]
        }
    }
)

result = agent.run("""
    1. Navigate to https://news.ycombinator.com
    2. Find the top 10 stories
    3. Save them as a JSON file called hn_top10.json with title, url, and score
""")
# Claude ישתמש ב-Playwright לגלישה ו-Write לשמירת הקובץ

MCP Tool Search — טעינה עצלנית

כשיש הרבה כלי MCP (50+), הם צורכים מקום ב-context. ה-SDK מפעיל אוטומטית deferred loading — כלים נטענים רק כשClaude צריך אותם. זה חוסך כ-85% מצריכת הטוקנים של הגדרות כלים (מ-77K ל-8.7K בדוגמה עם 50+ כלים).

MCP Servers מותאמים — חיבור ל-APIs שלכם

אפשר לכתוב MCP server מותאם שמחבר את הסוכן ל-APIs פנימיים, מסדי נתונים מותאמים, או שירותים ספציפיים לארגון שלכם. זה מרחיב את הסוכן בלי לשנות שורת קוד ב-SDK — רק מוסיפים server חדש.

דוגמאות ל-MCP servers שימושיים:

היופי: הסוכן לא צריך לדעת שה-MCP קיים מראש. הוא רואה את הכלים הזמינים ובוחר את המתאים. אם ביקשתם "check the database for users created today" — הוא ישתמש אוטומטית ב-PostgreSQL MCP אם זמין, או ינסה גישה אחרת אם לא.

טיפול בשגיאות וחוסן

סוכנים בפרודקשן חייבים לדעת להתמודד עם כשלונות. אם Pipeline שרץ כל לילה נתקע בגלל שגיאה — ואין טיפול — תגלו רק בבוקר שהנתונים לא עובדו. הנה מה שיכול להשתבש ואיך מטפלים:

קטגוריות שגיאות

Retry עם Exponential Backoff

ה-SDK כולל retry מובנה לשגיאות API זמניות (rate limits, 503). אבל אם אתם רוצים שליטה מלאה:

import time
from claude_agent_sdk import Agent, AgentError

def run_with_retry(agent, prompt, max_retries=3, base_delay=5):
    """הרצה עם retry — exponential backoff."""
    for attempt in range(max_retries):
        try:
            return agent.run(prompt)
        except AgentError as e:
            if attempt == max_retries - 1:
                raise  # ניסיון אחרון — תזרוק
            delay = base_delay * (2 ** attempt)  # 5, 10, 20 שניות
            print(f"Attempt {attempt + 1} failed: {e}")
            print(f"Retrying in {delay}s...")
            time.sleep(delay)

agent = Agent(model="claude-sonnet-4-6", max_turns=10)
result = run_with_retry(agent, "process the data files")

Fallback — מודל חלופי

def run_with_fallback(prompt, primary_model="claude-opus-4-6",
                      fallback_model="claude-sonnet-4-6"):
    """נסה Opus, אם נכשל — חזור ל-Sonnet."""
    try:
        agent = Agent(model=primary_model, max_turns=15)
        return agent.run(prompt)
    except AgentError as e:
        print(f"Primary ({primary_model}) failed: {e}")
        print(f"Falling back to {fallback_model}...")
        agent = Agent(model=fallback_model, max_turns=15)
        return agent.run(prompt)

Circuit Breaker — מנתק מעגל

דפוס חשוב: אם סוכן נכשל N פעמים ברציפות — תפסיקו לנסות ותתריעו לאדם. בלי זה, אתם שורפים טוקנים (= כסף) על משימה שבורה.

class CircuitBreaker:
    """מנתק מעגל — עוצר אחרי N כשלונות רצופים."""
    def __init__(self, max_failures=3):
        self.max_failures = max_failures
        self.consecutive_failures = 0

    def record_success(self):
        self.consecutive_failures = 0

    def record_failure(self):
        self.consecutive_failures += 1
        if self.consecutive_failures >= self.max_failures:
            raise RuntimeError(
                f"Circuit breaker: {self.consecutive_failures} "
                f"consecutive failures. Alerting human."
            )

# שימוש
breaker = CircuitBreaker(max_failures=3)
agent = Agent(model="claude-sonnet-4-6")

for task in daily_tasks:
    try:
        result = agent.run(task)
        breaker.record_success()
        process_result(result)
    except AgentError as e:
        breaker.record_failure()  # אחרי 3 — RuntimeError
        log_error(task, e)

דוגמה מלאה: Data Pipeline Agent

הגענו לדוגמה הראשונה מהעולם האמיתי. נבנה סוכן Python שרץ כל לילה ומעבד קבצי נתונים — Data Pipeline. הסוכן קורא CSVs מתיקיית inbox, מנקה ומוודא איכות, מייצר דו"ח, ומעביר קבצים מעובדים לארכיון. דמיינו חברת e-commerce ישראלית שמקבלת כל יום קבצי מלאי מ-3 ספקים שונים — כל אחד בפורמט קצת אחר, עם שגיאות בתאריכים ומחירים. הסוכן הזה מסדר את הבלגן.

הארכיטקטורה

data/
├── inbox/          # CSVs חדשים מגיעים לכאן
├── processed/      # קבצים שעובדו בהצלחה
├── quarantine/     # קבצים שנכשלו בעיבוד
└── reports/        # דו"חות איכות

הקוד המלא

"""
data_pipeline_agent.py — סוכן עיבוד נתונים ליצור
רץ כל לילה (cron), מעבד CSVs, מנקה, מדווח.
"""
import os
import json
from datetime import datetime
from claude_agent_sdk import Agent, AgentError

# --- הגדרות ---
INBOX_DIR = "data/inbox"
PROCESSED_DIR = "data/processed"
QUARANTINE_DIR = "data/quarantine"
REPORTS_DIR = "data/reports"
MODEL = "claude-sonnet-4-6"  # Sonnet מספיק — חוסך עלות
MAX_TURNS = 25               # מספיק לעיבוד 5-10 קבצים

def create_pipeline_agent():
    """יצירת סוכן עם הגדרות pipeline."""
    return Agent(
        model=MODEL,
        max_turns=MAX_TURNS,
        timeout=300,
        working_directory=os.getcwd(),
        allowed_paths=[INBOX_DIR, PROCESSED_DIR,
                       QUARANTINE_DIR, REPORTS_DIR]
    )

def run_pipeline():
    """הרצת ה-pipeline — הפונקציה הראשית."""
    agent = create_pipeline_agent()
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")

    prompt = f"""
    You are a data pipeline agent. Process CSV files in {INBOX_DIR}/.

    For EACH CSV file:
    1. READ the file and validate:
       - No empty required fields (first 3 columns must have values)
       - Dates are in YYYY-MM-DD format
       - Numbers are valid (no text in numeric columns)
       - No exact duplicate rows
    2. CLEAN issues you find:
       - Standardize date formats to YYYY-MM-DD
       - Trim whitespace from text fields
       - Remove exact duplicate rows
       - Replace obviously wrong values (negative prices) with null
    3. SAVE the cleaned file to {PROCESSED_DIR}/
       with the same name
    4. If a file is too corrupted to clean (>50% bad rows),
       move it to {QUARANTINE_DIR}/ instead

    After processing ALL files, write a quality report to:
    {REPORTS_DIR}/report_{timestamp}.md

    The report should include:
    - How many files processed successfully
    - How many quarantined
    - Per-file: rows before/after cleaning, issues found
    - Overall data quality score (0-100)

    Return a JSON summary with: files_processed, files_quarantined,
    total_rows_cleaned, quality_score.
    """

    try:
        result = agent.run(prompt)
        print(f"Pipeline completed. Cost: ${result.cost:.4f}")
        print(f"Tool calls: {len(result.tool_calls)}")
        print(f"\nSummary:\n{result.text}")

        # שמירת audit trail
        audit = {
            "timestamp": timestamp,
            "cost": result.cost,
            "tool_calls": len(result.tool_calls),
            "tokens": {
                "input": result.tokens_used.input,
                "output": result.tokens_used.output
            },
            "summary": result.text
        }
        with open(f"{REPORTS_DIR}/audit_{timestamp}.json", "w") as f:
            json.dump(audit, f, indent=2, ensure_ascii=False)

    except AgentError as e:
        print(f"Pipeline failed: {e}")
        # כאן הייתם שולחים התראה — Slack, email, PagerDuty
        raise

if __name__ == "__main__":
    # ודאו שתיקיות קיימות
    for d in [INBOX_DIR, PROCESSED_DIR, QUARANTINE_DIR, REPORTS_DIR]:
        os.makedirs(d, exist_ok=True)
    run_pipeline()

מה הסוכן עושה בפועל?

בואו נעקוב אחרי ריצה טיפוסית. נניח שב-data/inbox/ יש 3 קבצים:

1. sales_march.csv (200 שורות, נקי) — Claude קורא, בודק, הכל תקין → מעביר ל-processed/
2. customers_update.csv (150 שורות, 20 עם תאריכים בפורמט DD/MM/YYYY במקום YYYY-MM-DD) — Claude מזהה, מתקן את הפורמט, מסיר 3 כפילויות → מעביר ל-processed/ עם 147 שורות נקיות
3. broken_export.csv (100 שורות, 70 חסרות עמודות חובה) — 70% פגומות, מעל הסף של 50% → מעביר ל-quarantine/

הדו"ח שנוצר ב-reports/ יכיל סיכום per-file: כמה שורות לפני ואחרי ניקוי, אילו בעיות נמצאו, ציון איכות כולל. ב-audit_*.json תראו: 12 tool calls (Glob + 3xRead + 2xWrite + Bash למעבר קבצים + Write לדו"ח), עלות $0.35, 45K input tokens.

תזמון עם Cron

# crontab -e
# הרצה כל לילה ב-02:00
0 2 * * * cd /home/user/my-project && /home/user/agent-env/bin/python data_pipeline_agent.py >> /var/log/pipeline.log 2>&1

טיפ חשוב: ודאו שה-virtual environment מופעל ב-cron — השתמשו בנתיב מלא ל-Python (/home/user/agent-env/bin/python) ולא רק python3. כמו כן, ANTHROPIC_API_KEY צריך להיות זמין — הוסיפו source /home/user/.claude-env.sh && לפני הפקודה, או הגדירו ישירות ב-crontab.

עלות לריצה

דוגמה מלאה: Monitoring Agent

הדוגמה השנייה מהעולם האמיתי: סוכן שמנטר מערכות ושולח התראות חכמות. ההבדל מ-monitoring מסורתי (Prometheus, Datadog): כלי מסורתי מתריע על ספים (CPU > 90%). Claude מבין הקשר ומפחית false positives.

היתרון של Monitoring חכם

Monitoring מסורתי: "CPU reached 95%". Claude: "CPU reached 95% because a scheduled backup started at 02:00 — this happens every night and resolves in 20 minutes. No action needed." ההבדל: במקום להעיר מישהו ב-2 בלילה, Claude מבין שזה תקין.

הקוד

"""
monitoring_agent.py — סוכן ניטור חכם
בודק לוגים, API health, ומדדים. מתריע רק על בעיות אמיתיות.
"""
import os
import json
import requests
from datetime import datetime
from claude_agent_sdk import Agent, AgentError

# --- הגדרות ---
LOG_FILE = "/var/log/app/app.log"
HEALTH_ENDPOINTS = [
    "https://api.myapp.com/health",
    "https://api.myapp.com/db/health",
]
ALERT_WEBHOOK = os.environ.get("SLACK_WEBHOOK_URL", "")
MODEL = "claude-sonnet-4-6"  # Sonnet לבדיקות שגרתיות

def create_monitor_agent():
    """יצירת סוכן monitor."""
    return Agent(
        model=MODEL,
        max_turns=15,
        timeout=120,
        allowed_tools=["Read", "Bash", "Grep", "Glob", "WebFetch"]
        # אין Write/Edit — monitor לא צריך לשנות קבצים
    )

def check_health_endpoints():
    """בדיקת health endpoints ידנית — מהירה וזולה."""
    results = {}
    for url in HEALTH_ENDPOINTS:
        try:
            resp = requests.get(url, timeout=10)
            results[url] = {
                "status": resp.status_code,
                "latency_ms": resp.elapsed.total_seconds() * 1000,
                "ok": resp.status_code == 200
            }
        except requests.RequestException as e:
            results[url] = {"status": 0, "error": str(e), "ok": False}
    return results

def send_alert(message):
    """שליחת התראה ל-Slack."""
    if ALERT_WEBHOOK:
        requests.post(ALERT_WEBHOOK, json={"text": message})
    print(f"ALERT: {message}")

def run_monitor():
    """הרצת סקר ניטור."""
    agent = create_monitor_agent()
    health = check_health_endpoints()
    timestamp = datetime.now().isoformat()

    prompt = f"""
    You are a monitoring agent. Analyze system health.

    Current time: {timestamp}

    ## Health Check Results:
    {json.dumps(health, indent=2)}

    ## Tasks:
    1. Read the last 200 lines of {LOG_FILE}
    2. Look for: ERROR, CRITICAL, Exception, Timeout patterns
    3. Correlate: do the log errors relate to the health check failures?

    ## Analysis Rules:
    - Scheduled tasks (backup at 02:00, cleanup at 03:00) are NORMAL
    - Occasional 503s during deployments are NORMAL
    - 3+ consecutive errors on the same endpoint = PROBLEM
    - Error rate > 5% in last 100 lines = PROBLEM
    - Any CRITICAL log entry = ALWAYS ALERT

    ## Output Format:
    Return JSON:
    {{
        "status": "ok" | "warning" | "critical",
        "summary": "one sentence",
        "details": "detailed analysis",
        "action_needed": true/false,
        "alert_message": "message for humans (if action needed)"
    }}
    """

    try:
        result = agent.run(prompt)
        print(f"Monitor completed. Cost: ${result.cost:.4f}")

        # ניסיון לפרסר JSON מהתוצאה
        try:
            analysis = json.loads(result.text)
            if analysis.get("action_needed"):
                send_alert(analysis.get("alert_message", "Monitor alert"))
            return analysis
        except json.JSONDecodeError:
            # Claude לא החזיר JSON נקי — נשתמש בטקסט
            if "critical" in result.text.lower():
                send_alert(f"Monitor found issues: {result.text[:500]}")
            return {"status": "unknown", "raw": result.text}

    except AgentError as e:
        send_alert(f"Monitor agent failed: {e}")
        return {"status": "error", "error": str(e)}

if __name__ == "__main__":
    result = run_monitor()
    print(json.dumps(result, indent=2, ensure_ascii=False))

אופטימיזציית עלויות ל-Monitoring

Monitoring רץ הרבה — כל 5 דקות, כל שעה. זה מצטבר. הנה אסטרטגיה:

הטריק: בדיקות בסיסיות (health endpoints) רצות בקוד Python רגיל — בלי Claude, בלי עלות טוקנים. Claude מופעל רק כשצריך אינטליגנציה: ניתוח לוגים, קורלציה, או כשנמצאה חריגה שדורשת הבנה. בתכלס, זה כמו לשכור שומר ראש: רוב הזמן המצלמות עושות את העבודה, אבל כשמשהו חשוד — השומר בודק.

Escalation — שדרוג מודל לפי חומרה

אסטרטגיה מתקדמת: השתמשו ב-Sonnet לבדיקות שגרתיות. כש-Sonnet מזהה משהו חשוד — העבירו ל-Opus לניתוח מעמיק:

def smart_monitor():
    """Sonnet לשגרה, Opus כשיש חריגה."""
    # שלב 1: סריקה מהירה עם Sonnet
    quick_agent = Agent(model="claude-sonnet-4-6", max_turns=10)
    scan = quick_agent.run(f"Quick scan of {LOG_FILE} — any anomalies?")

    if "anomaly" in scan.text.lower() or "critical" in scan.text.lower():
        # שלב 2: ניתוח מעמיק עם Opus
        deep_agent = Agent(model="claude-opus-4-6", max_turns=20)
        analysis = deep_agent.run(
            f"Deep analysis needed. Previous scan found: {scan.text}\n"
            f"Investigate {LOG_FILE} thoroughly.",
            messages=scan.messages  # Context מהסריקה הראשונה
        )
        return analysis

    return scan  # הכל תקין — Sonnet מספיק

התוצאה: 95% מהריצות עולות $0.01 (Sonnet שגרתי). רק 5% מגיעות ל-Opus ($0.50). עלות ממוצעת: $0.035 לבדיקה — שליש מהעלות של Opus על הכל.

טסטים לסוכני SDK

סוכנים הם לא-דטרמיניסטיים. שתי ריצות עם אותו prompt יכולות לתת תוצאות שונות. זה אתגר לטסטים. הנה הפיתרון: פירמידת הטסטים לסוכנים.

Mock Tests — בלי API, בלי עלות

"""test_pipeline.py — טסטים עם mocks"""
import unittest
from unittest.mock import patch, MagicMock

class TestPipelineAgent(unittest.TestCase):

    @patch('claude_agent_sdk.Agent')
    def test_pipeline_processes_clean_csv(self, MockAgent):
        """CSV נקי צריך להגיע ל-processed/."""
        # Mock — מדמה תוצאה מוצלחת
        mock_result = MagicMock()
        mock_result.text = '{"files_processed": 1, "files_quarantined": 0}'
        mock_result.cost = 0.05
        mock_result.tool_calls = [MagicMock(tool="Read"), MagicMock(tool="Write")]
        MockAgent.return_value.run.return_value = mock_result

        # הפעלה
        from data_pipeline_agent import run_pipeline
        # בדיקה שהסוכן נקרא עם prompt שכולל "inbox"
        MockAgent.return_value.run.assert_called_once()
        call_args = MockAgent.return_value.run.call_args
        self.assertIn("inbox", call_args[0][0])

    @patch('claude_agent_sdk.Agent')
    def test_pipeline_handles_agent_error(self, MockAgent):
        """כשהסוכן נכשל — הקוד צריך לטפל."""
        from claude_agent_sdk import AgentError
        MockAgent.return_value.run.side_effect = AgentError("API down")

        from data_pipeline_agent import run_pipeline
        with self.assertRaises(AgentError):
            run_pipeline()

if __name__ == '__main__':
    unittest.main()

Integration Tests — עם API אמיתי, על סביבת ניסוי

"""test_pipeline_integration.py — אינטגרציה אמיתית (עולה טוקנים!)"""
import os
import json
import unittest

class TestPipelineIntegration(unittest.TestCase):
    """רצים רק עם INTEGRATION=1 — כי עולים כסף."""

    @unittest.skipUnless(os.environ.get("INTEGRATION"), "Integration tests disabled")
    def test_real_pipeline_run(self):
        """ריצה אמיתית על קבצי ניסוי."""
        from data_pipeline_agent import run_pipeline
        # ... הכנת קבצי ניסוי ...
        run_pipeline()
        # בדיקה שקבצים עובדו
        self.assertTrue(os.path.exists("data/processed/test.csv"))

    @unittest.skipUnless(os.environ.get("INTEGRATION"), "Integration tests disabled")
    def test_golden_output(self):
        """השוואה ל-golden output — מבנה, לא טקסט מדויק."""
        from claude_agent_sdk import Agent
        agent = Agent(model="claude-sonnet-4-6", max_turns=5)
        result = agent.run("list all .py files and count them")

        # לא בודקים טקסט מדויק (לא דטרמיניסטי)
        # בודקים מבנה ו-intent
        self.assertIn(".py", result.text)
        self.assertTrue(any(c.tool == "Glob" for c in result.tool_calls))
        self.assertGreater(result.cost, 0)

# הרצה: INTEGRATION=1 python -m pytest test_pipeline_integration.py

Golden Output — בדיקה לפי תבנית

הדפוס הכי שימושי לטסטים של סוכנים: Golden Output. הרעיון: הריצו את הסוכן פעם אחת, בדקו שהתוצאה נכונה, שמרו את המבנה (לא הטקסט המדויק). בטסטים הבאים — השוו מבנה.

# שמירת golden output (פעם אחת, ידנית)
golden = {
    "expected_tools": ["Glob", "Read"],      # כלים שחייבים להופיע
    "expected_keywords": [".py", "function"], # מילות מפתח בתוצאה
    "max_cost": 0.10,                         # סף עלות סביר
    "min_tool_calls": 2,                      # מינימום פעולות
}

# בטסט — השוואה למבנה
def test_agent_matches_golden():
    result = agent.run("list Python files and their functions")
    used_tools = {c.tool for c in result.tool_calls}

    assert all(t in used_tools for t in golden["expected_tools"])
    assert all(k in result.text for k in golden["expected_keywords"])
    assert result.cost <= golden["max_cost"]
    assert len(result.tool_calls) >= golden["min_tool_calls"]

היתרון: הטסט עובר גם אם Claude מנסח את התשובה קצת אחרת. מה שמשנה הוא שהוא עשה את הדברים הנכונים (השתמש בכלים הנכונים) והחזיר תוכן רלוונטי (מילות מפתח).