יצירת MCP Server כדי לבצע פעולות שונות עם LLM

הפוסט הזה הוא המשך ישיר לפוסט הקודם של מבוא ל-MCP – דרך מצוינת לחבר LLM לשירותים שונים. בפוסט הזה אנחנו מסבירים איך יוצרים MCP Server משלנו כדי ליצור פעולות שונות.

למה אני צריך MCP Server משלי?

בגדול, יש רשימה עצומה של MCP Servers שונים שמתחברים ומתממשקים לשירותים שונים ומשונים: מגיטהאב ועד דברים לוקליים. זה מעולה, אבל אנחנו צריכים לעתים MCP שהוא של ה… החברה שלנו או העסק שלנו. למשל MCP שיפעיל שירות ספציפי של החברה שלי או יקבל ממנה מידע לאחר אותנטיקציה. אם אני עובד בחברת הדפסה למשל, אז שה-MCP שלי ישלח הוראות לדפוס, אלבום לדפוס או ישלח מידע לשרת כדי לקבל מידע שרק החברה שלי יודעת לספק או לתת.

אפשר לומר ש-MCP Server זו האבולוציה של ההתממשקות של העסקים בעידן ה-LLM.

הדוגמה שלי – גישה לאינטרנט ישראל

אני מאד אוהב להרגיש ולנסות דברים עם הידיים, אז אני אדגים ואכתוב פה עם משהו שקל להבין – MCP Server ייחודי לי שמביא את ה-RSS האחרון של האתר שלי. זה לא שימושי בכלל, כי בינינו ה-LLMים האחרונים יודעים כבר לגשת אוטומטית לפידים האלו ואני לא חוסם אותם (אפילו שאני יכול!) אבל לצורך הדוגמה והשעשוע, אני אעשה את זה.

איך ניגשים ל-RSS?

החלק הזה לא רלוונטי ל-MCP אבל כן רלוונטי למה שאני רוצה לעשות עם MCP. אני מציע לכם לבודד את החלק הפונקציונלי שעושה את הפעולה/מביא את המידע שה-MCP צריך לבין החלק שאחראי ל-MCP.

עבור החלק הפונקציוןנלי אנו ניצור פונקציה בפייתון שמשתמשים בה כדי לגשת, לפרסר ולהציג RSS. אני מעדיף לבודד את הפונקציונליות הזו ולבדוק אותה במנותק מה-LLM, אז אני אשמור את הקוד הזה בקובץ פייתון נפרד בשם test-rss.py.

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

uv init

אני אתקין שתי חבילות שנדרשות לקריאת RSS – הראשונה היא httpx שמבצעת את הקריאה לשרת מרוחק ו-feedparser שמפרסרת את ה-XML של ה-RSS. עם uv זה די קל:

uv add httpx feedparser

ועכשיו ליצירת הקוד. טוב, פה ה-LLM יכול לעזור. מה שטוב – אם אני מפריד את החלק הפונקציונלי מהחלק של ה-MCP, מאד קל לי להבין אם הקוד שלי עובד. במקרה הזה – מביא לי את חמשת הפוסטים האחרונים.

import httpx
import feedparser

FEED_URL = "https://internet-israel.com/feed"

async def fetch_rss(url: str) -> str | None:
    """Fetch raw RSS feed content asynchronously."""
    try:
        headers = {
            "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36",
            "Accept": "application/rss+xml,application/xml;q=0.9,*/*;q=0.8",
        }

        async with httpx.AsyncClient(follow_redirects=True) as client:
            response = await client.get(url, headers=headers, timeout=10)
            if response.status_code == 200:
                return response.text
            else:
                print(f" HTTP error: {response.status_code}")
    except Exception as e:
        print(f" Exception occurred: {e}")
    return None


async def main():
    raw_xml = await fetch_rss(FEED_URL)
    if not raw_xml:
        print(" Failed to fetch the feed.")
        return

    feed = feedparser.parse(raw_xml)
    if not feed.entries:
        print(" No entries found.")
        return

    for entry in feed.entries[:5]:
        print(f"\n {entry.title}\n {entry.get('published', 'N/A')}\n {entry.link}")

# Run the async function directly
if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

אני יכול להריץ את קוד הפייתון הזה בנפרד כדי לראות אם הוא בסדר. אני אכלתי קצת קש כי בקוד המקורי לא היה User-Agent ואז הקלאודפלייר בלם את הסורק האוטומטי.

שרת ה-MCP

עכשיו, כשיש לי חלק פונקציונלי שעושה פעולה מסוימת, אפשר לממש את שרת ה-MCP. נוריד את החבילה mcp[cli] שהיא החבילה הרשמית של אנטרופיק:

uv add "mcp[cli]"

השלב הבא הוא בדרך כלל להתחיל לכתוב ולבקש מהקופיילוט או מה-LLM שיש לכם להתחיל לכתוב את הקוד, הבעיה היא ש-MCP הוא חדש ואם תבקשו – לפחות בשעת כתיבת שורות אלו, תקבלו הזיות שאין דברים כאלו. מה עושים? יש בדוקומנטציה של אנתרופיק דוגמה מוצלחת. מעתיקים אותה ומבקשים מה-LLM שיעבוד לפי הדוגמה הזו. או מעתיקים את הקוד שלי (אחרי כל ההסברים) ומבקשים את הדוגמה הזו.

הקוד שלי הוא פשוט למדי וכולל גם את הקוד הפונקציונלי וגם את הקוד שיוצר את ה-MCP שהוא בערך שלוש שורות. אז קודם כל – הנה הקוד המלא ואחר כך אסביר.

from typing import Optional
import httpx
import feedparser
from mcp.server.fastmcp import FastMCP

# Initialize FastMCP
mcp = FastMCP("internet_israel_feed")

FEED_URL = "https://internet-israel.com/feed/"  # Use HTTPS

async def fetch_rss(url: str) -> str | None:
    """Fetch raw RSS feed content asynchronously."""
    try:
        headers = {
            "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36",
            "Accept": "application/rss+xml,application/xml;q=0.9,*/*;q=0.8",
        }

        async with httpx.AsyncClient(follow_redirects=True) as client:
            response = await client.get(url, headers=headers, timeout=10)
            if response.status_code == 200:
                return response.text
            else:
                print(f" HTTP error: {response.status_code}")
    except Exception as e:
        print(f" Exception occurred: {e}")
    return None

@mcp.tool()
async def get_feed(limit: int = 5) -> str:
    """Fetch latest blog posts from internet-israel.com

    Args:
        limit: Number of feed items to return (default 5)
    """
    raw_feed = await fetch_rss(FEED_URL)
    if not raw_feed:
        return " Failed to fetch the RSS feed."

    feed = feedparser.parse(raw_feed)
    if not feed.entries:
        return " No entries found in the feed."

    entries = []
    for entry in feed.entries[:limit]:
        formatted = f"""
 {entry.title}
 {entry.get('published', 'No date')}
 {entry.link}
""".strip()
        entries.append(formatted)

    return "\n\n---\n\n".join(entries)
# Run the server
if __name__ == "__main__":
    mcp.run(transport="stdio")

חלק גדול מהקוד הוא הקוד הפונקציונלי, זה שעושה דברים. שרת ה-MCP הוא:

from mcp.server.fastmcp import FastMCP

כאן יש את הייבוא של המודול שמריץ את השרת.

mcp = FastMCP("internet_israel_feed")

הקוד שיוצר את השרת וגם כולל את השם שלו – זה חלק מאד מאד מאד חשוב. כי את השם אנו נצטרך בחיבור ל-MCP קליינט. אנחנו מקבלים רפרנס שנשתמש בו אחר כך.

@mcp.tool()

זה דקורטור שבא לפני הפקודה עצמה. שם הפונקציה הוא בעצם שם הפקודה. במקרה הזה הצבתי את הדקורטור לפני פונקציה get_feed אז get_feed יהיה שם הפקודה שלי.

mcp.run(transport="stdio")

ולבסוף, הרצת השרת! אני אריץ אותו עם uv run main.py או כל שם קובץ אחר שבחרתי לקובץ הפייתון. חשוב לבדוק שהוא עולה ורץ. אפשר להוריד אותו ואז… אני אגש אל הקליינט.

להוסיף את השרת שלנו לקליינט

במאמר הקודם עשינו שימוש בקונפיגורציה של mcp-cli להוספת שרת שכבר קיים. עכשיו אנחנו צריכים להוסיף את השרת שלנו. זה כבר די קל –

      "internet-israel-rss": {
        "command": "uv",
        "args": ["run", "--directory", "/Users/barzik/local/demos/mcp-server", "internet-israel-rss.py"]
      }

מאד חשוב לשים לב לכתוב את ה-run ראשון ברשימת הפקודות ולהחליף את הנתיב לנתיב שלכם (אלא אם כן גם שם המשתמש שלכם barzik!). אם הכל עובד, כשתפעילו את ה-mcp cli תקבלו חיווי שעלה לכם כלי אחד. אם תעשו tools/ אז תוכלו לראות את רשימת הכלים. במקרה הזה כלי אחד ופקודה אחת. אפשר גם להפעיל את הפקודות ישירות כדי לראות שהכל עובד.

זה המקום שבו אפשר לדבג דברים. אבל אם הקפדתם שתהיה הפרדה בין הקוד הפונקציונלי לקוד של ה-MCP server, לא תצטרכו להתקל בבעיות פונקציונליות של (למשל) – אני לא מצליח לגשת לשרת חיצוני, יש לי בעיה בקוד שקורא ל-RSS או בפירסור הפלט הזה.

כמובן שאפשר גם לבדוק שימוש במענה ישיר של ה-LLM. למשל בשאלה בשפה טבעית.

יש כמובן עוד המון ניואנסים שנכנס אליהם בפוסטים הבאים בסדרה, אבל אפשר לראות עד כמה זה קל יחסית. כל מה שצריך זה תכנון נכון ומחשבה וזה באמת… פשוט ביותר ליצור איזה פעולות שאתם רוצים ולחבר אותם ל-LLM.