יצירת mcp client

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

אני מדגים עם פייתון, אבל ממש אין שום בעיה לעבוד עם Node.js או כל שפה אחרת (כן, כן, גם #C) כדי ליצור קליינט. החבילות של MCP זמינות לכל השפות.

התחלת הפרויקט

אנו נתחיל מליצור פרויקט ב-uv. אם אתם לא מכירים uv, אז באמת כדאי להכיר ובפוסט הזה אני מסביר. לא סתם הוא בוער ובוער חזק בתחום פיתוח הפייתון. נפתח תיקיה, נכנס אליה, נקליד uv init כדי ליצור פרויקט uv ואז uv venv כדי ליצור סביבה וירטואלית. אנחנו מוכנים. נתקין את המודולים של mcp וכיוון שאני רוצה להשתמש ב-LLM של openai, אז נשתמש בו. אבל הוא באמת לא חובה עבור הקליינט, כן?

uv add openai mcp

החלטה לאיזה שרת mcp להתחבר

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

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

אם אתם רוצים, אתם יכולים להריץ את הקוד הזה. צריך API key של openAI. יצרו אחד עם החשבון שלכם והכניסו אותו כמשתנה סביבה עם:

export OPENAI_API_KEY="sk-proj-YOUR-KEY"

העתיקו את הקוד הזה לקובץ בשם client-stdio.py:

import asyncio
import os
import sys
from typing import Optional
from contextlib import AsyncExitStack

from openai import OpenAI

# MCP imports for stdio communication
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

class MCPStdioClient:
    """
    An MCP client that connects to an MCP server using stdio communication
    and processes queries using OpenAI and the server's tools.
    """
    
    def __init__(self):
        # Initialize OpenAI client
        self.openai = OpenAI()
        
        # Initialize session and stdio objects
        self.session: Optional[ClientSession] = None
        self.stdio = None
        self.write = None
        self.exit_stack = AsyncExitStack()
        
        # Store server tools
        self.tools = []

    async def connect_to_server(self, server_script_path: str, use_uv: bool = False, server_dir: str = None) -> bool:
        """
        Connect to an MCP server via stdio
        
        Args:
            server_script_path: Path to the server script (.py or .js)
            use_uv: Whether to use uv to run the server
            server_dir: Directory to run uv from
            
        Returns:
            bool: True if connection successful, False otherwise
        """
        try:
            print(f"Connecting to MCP server at: {server_script_path}")
            
            # Determine server type (Python or Node.js)
            is_python = server_script_path.endswith('.py')
            is_js = server_script_path.endswith('.js')
            if not (is_python or is_js):
                print("Error: Server script must be a .py or .js file")
                return False

            # Configure server parameters based on whether we're using uv or not
            env = os.environ.copy()  # Use current environment with OPENAI_API_KEY
            
            if use_uv and is_python and server_dir:
                # Using uv to run the server from specified directory
                print(f"Using uv to run server from directory: {server_dir}")
                command = "uv"
                script_name = os.path.basename(server_script_path)
                args = ["--directory", server_dir, "run", script_name]
            else:
                # Using default Python or Node interpreter
                command = "python" if is_python else "node"
                args = [server_script_path]

            server_params = StdioServerParameters(
                command=command,
                args=args,
                env=env
            )

            # Connect to server
            stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))
            self.stdio, self.write = stdio_transport
            self.session = await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))

            # Initialize session
            print("Initializing MCP session...")
            await self.session.initialize()

            # Fetch available tools
            response = await self.session.list_tools()
            self.tools = response.tools
            
            if self.tools:
                print(f"\nConnected to server with {len(self.tools)} tools:")
                for tool in self.tools:
                    print(f"  - {tool.name}: {tool.description}")
            else:
                print("Connected to server but no tools available")
                
            return True
            
        except Exception as e:
            print(f"Error connecting to MCP server: {str(e)}")
            import traceback
            traceback.print_exc()
            return False

    async def process_query(self, query: str) -> str:
        """
        Process a query using OpenAI and the MCP server's tools
        
        Args:
            query: The user's question or request
            
        Returns:
            OpenAI's response as a string
        """
        if not self.session:
            return "Error: Not connected to an MCP server"
            
        try:
            # Prepare initial chat messages
            messages = [
                {"role": "system", "content": "You are a helpful assistant that can access the latest blog posts from internet-israel.com."},
                {"role": "user", "content": query}
            ]

            # Convert MCP tools to OpenAI tool format
            available_tools = [{
                "type": "function",
                "function": {
                    "name": tool.name,
                    "description": tool.description,
                    "parameters": tool.inputSchema
                }
            } for tool in self.tools]

            # Initial OpenAI API call with tools
            print("Sending query to OpenAI...")
            response = self.openai.chat.completions.create(
                model="gpt-4-turbo",  # Using GPT-4 Turbo for tool use
                messages=messages,
                tools=available_tools,
                tool_choice="auto",  # Let the model decide whether to use tools
                max_tokens=1000
            )

            # Process response and handle tool calls
            final_text = []
            
            assistant_message = response.choices[0].message
            if assistant_message.content:
                final_text.append(assistant_message.content)
            
            # Handle tool calls if present
            if assistant_message.tool_calls:
                # Add assistant message to conversation
                messages.append(assistant_message)
                
                print(f"OpenAI wants to use {len(assistant_message.tool_calls)} tools")
                
                # Process each tool call
                for tool_call in assistant_message.tool_calls:
                    tool_name = tool_call.function.name
                    tool_args = tool_call.function.arguments
                    
                    # Convert string arguments to JSON if needed
                    import json
                    if isinstance(tool_args, str):
                        try:
                            tool_args = json.loads(tool_args)
                        except json.JSONDecodeError:
                            print(f"Warning: Could not parse tool args as JSON: {tool_args}")
                            tool_args = {}
                    
                    print(f"Calling tool: {tool_name} with args: {tool_args}")
                    
                    # Execute tool call through MCP server
                    try:
                        result = await self.session.call_tool(tool_name, tool_args)
                        tool_result = result.content
                        final_text.append(f"\n[Tool: {tool_name}]\n{tool_result}")
                        
                        # Add tool result to messages
                        messages.append({
                            "role": "tool",
                            "tool_call_id": tool_call.id,
                            "content": tool_result
                        })
                    except Exception as e:
                        error_msg = f"Error calling tool {tool_name}: {str(e)}"
                        print(error_msg)
                        final_text.append(f"\n[Error with tool {tool_name}]: {str(e)}")
                        
                        # Add error to messages
                        messages.append({
                            "role": "tool",
                            "tool_call_id": tool_call.id,
                            "content": f"Error: {str(e)}"
                        })
                
                # Get final response from OpenAI with tool results
                print("Getting final response from OpenAI with tool results...")
                response = self.openai.chat.completions.create(
                    model="gpt-4-turbo",
                    messages=messages,
                    max_tokens=1000
                )
                
                next_message = response.choices[0].message
                if next_message.content:
                    final_text.append(f"\n{next_message.content}")

            return "\n".join([text for text in final_text if text])
            
        except Exception as e:
            print(f"Error processing query: {str(e)}")
            import traceback
            traceback.print_exc()
            return f"Error processing query: {str(e)}"

    async def chat_loop(self):
        """Run an interactive chat loop"""
        print("\nMCP Stdio Client Started!")
        print("Type your queries about internet-israel.com blog or 'quit' to exit.")

        while True:
            try:
                query = input("\nQuery: ").strip()

                if query.lower() in ('quit', 'exit'):
                    break

                response = await self.process_query(query)
                print("\nResponse:")
                print(response)

            except Exception as e:
                print(f"\nError: {str(e)}")
                import traceback
                traceback.print_exc()

    async def cleanup(self):
        """Clean up resources"""
        if self.session:
            await self.exit_stack.aclose()
            print("Connection to MCP server closed")


async def main():
    # Validate command line arguments
    if len(sys.argv) < 2:
        print("Usage: uv run client-stdio.py <path_to_mcp_server_script> [--use-uv] [--server-dir=DIR]")
        return
    
    server_script = sys.argv[1]
    
    # Parse additional command line arguments
    use_uv = "--use-uv" in sys.argv
    server_dir = None
    
    for arg in sys.argv:
        if arg.startswith("--server-dir="):
            server_dir = arg.split("=", 1)[1]
    
    # If using uv but no server directory specified, extract from server script path
    if use_uv and not server_dir:
        server_dir = os.path.dirname(server_script)
        if not server_dir:
            server_dir = "."
    
    # Initialize and run client
    client = MCPStdioClient()
    
    try:
        # Connect to server
        if await client.connect_to_server(server_script, use_uv, server_dir):
            # Start interactive chat loop
            await client.chat_loop()
    finally:
        # Clean up resources
        await client.cleanup()


if __name__ == "__main__":
    asyncio.run(main())

תדאגו שיהיה לכם mcp server מוכן לפעולה. אני השתמשתי ב-mcp server שהדגמתי במאמר הקודם. ה-mcp server הקצת מסכן הזה מכיל כלי אחד שמביא את 5 הפוסטים האחרונים של הבלוג הזה. אבל הוא מעולה כדוגמה. אני מפעיל אותו באמצעות uv דרך הקוד הזה:

uv --directory /Users/spare10/local/demo-mcp-server run main.py

אני לא נדרש להפעיל אותו, רק לזכור איך להפעיל אותו. כשאני קורא ל-mcp client שלי, אני צריך להעביר לו את רשימת השרתים שהוא מחובר אליהם. כן, בדיוק כמו ה-mcp cli שהוא בסופו של דבר גם קליינט. רק אחד שבנוי יותר טוב מהקקמייקה בן 150 השורות שנתתי לעיל. איך אני מפעיל את השרת? מהתיקיה שיש בה את הקובץ:

uv run client-stdio.py /Users/spare10/local/demo-mcp-server/main.py --use-uv

אם לא שכחתי והצבתי את ה-OPENAI_API_KEY כמשתנה סביבה, אני אראה שיש לי קליינט שגם יש בו את הכלי של get_feed!

או קיי – איך זה עובד?

איך הנס הזה קרה? כאמור הקליינט מחבר את ה-LLM לרשימת שרתים. במקרה הזה שרתי stdio. הקליינט נוצר באמצעות

client = MCPStdioClient()

ואז:

if await client.connect_to_server(server_script, use_uv, server_dir):

וזהו, ה-mcp client מפעיל את השרתים ויחד עם ה-LLM הוא מאפשר את הכל. עכשיו אנחנו בעצם יכולים לעבוד על הקליינט שלנו לפי הצרכים שלנו. פה בניתי דוגמה פשוטה שעובדת עם ה-CLI. כמובן שאנחנו לא חייבים לעשות כן. כל אחד והקליינט שלו. ברוב הפעמים זה יהיה קליינט שבכלל לא מדבר עם משתמש אלא עם קלט שמגיע ממקומות אחרים.

אני לא רוצה לנתח את הקוד כי כבר זה פחות מעניין בעולם של היום. מה שמעניין במיוחד מבחינתנו הוא לראות איך הקוד עובד. יצירת הקליינט, קבלת המידע על השרתים וחיבור של הכל באמצעות ה-SDK של ה-mcp. ה-MCPStdioClient יודע בעצם להתחבר לשרתים ולחבר בינם לבין פעולות ולדעת להפעיל אותם לפי הוראותיו של ה-LLM. מה שחשוב הוא למצוא דוגמה טובה שעובדת, לתת אותה ל-LLM שכותב לכם את הקוד ולומר לו ״תעשה ככה״ ואז לעבוד על הקוד שלו כדי לראות שהוא לא עושה שטויות ולהבין באמת אם המידע עובר ולאן.

כניסה אל מאחורי הקלעים

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

from openai import OpenAI

client = OpenAI()

tools = [{
    "type": "function",
    "name": "get_feed",
    "description": "Get last posts of internet-israel.com.",
    "parameters": {
        "type": "object",
        "properties": {
            "number": {
                "type": "int",
                "description": "Number of posts required"
            }
        },
        "additionalProperties": False
    }
}]

response = client.responses.create(
    model="gpt-4.1",
    input=[{"role": "user", "content": "What are the last post in internet israel?"}],
    tools=tools
)

print(response.output)

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

[{
    "type": "function_call",
    "id": "fc_12345xyz",
    "call_id": "call_12345xyz",
    "name": "get_feed",
    "arguments": "{\"number\":1}"
}]

האחריות שלי מי שקורה היא לקרוא לכלי ולספק את התשובה ואז לחכות לתשובה נוספת מה-LLM. עד ה-mcp היינו צריכים לעשות את זה ידנית ובאופן שונה מ-LLM אחד ל-LLM אחר. כי כל LLM עובד קצת אחרת. אז הייתי קורא ל-openAI ושולח לו את ה-tools בפורמט שהוא מבקש, לקבל את הפעלת ה-tools בפורמט שלו ואז להריץ אותם ולשלוח לו את התשובה. דבר שבאמת מצריך מאמץ וקוד למימוש כל הסיפור הזה. ואם משתנה גרסה או API או חלילה וחס אני רוצה להשתמש ב-LLM אחר? בהצלחה לי.

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

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

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