במאמרים הקודמים למדנו על MCP באופן כללי, המשכנו ל-MCP Server ואז ל-MCP Client. יש צורך בהכרות עם המאמרים הקודמים ורקע בתכנות כדי להבין את המאמר הזה שעוסק בדרך אחרת שבה קליינט של MCP יכול לתקשר עם שרת MCP – והיא Streamable HTTP Transport. מדובר בדרך חדשה שמחליפה דרך קודמת שנקרא SSE Transport.
עד כה, כל הדוגמאות המגניבות שהראנו של חיבור בין שרת לקליינט רצו על שרת מקומי. כלומר הפעלתי את השרת על המחשב והפעלתי את הקליינט על אותו מחשב. זה שימושי אם יש לי מכונה אחת שבה יש את הלקוח והשרת. למשל אם אני עושה דברים לוקלית זה נהדר. אבל כשאנו עובדים עם סקייל, למשל ארגוני, אנחנו רוצים להשתמש ב-MCP מרוחקים.
אני אתן דוגמה מהעולם שלי כלקוח. MCP מקומי מאפשר ל-LLM שלי לעשות שינויים על הקבצים המקומיים (למשל לערוך את הקוד), אבל MCP מרוחק מאפשר ל-LLM ולי לפתוח טיקטים לג׳ירה. איך? הג׳ירה מפעיל MCP מרוחק, אני מפעיל את הקליינט שניגש (עם הטוקן שלי) אל הג׳ירה ומבצע את ה-actions שהוא מציע.
כלומר אני מפעיל קליינט מקומי על המחשב שלי ומבקש ממנו את הבקשה הבאה: ״קרא את הטיקט האחרון בפרויקט בג׳ירה (מערכת לניהול באגים למי שמכיר), ותבצע את המשימה שיש שם״. הקליינט מחובר ל-MCP Server מקומי, שיכול לעשות שינוי בקבצים אבל… איך הוא יגש לג׳ירה? במקרה שלנו, הוא יצטרך לגשת, באמצעות האינטרנט, ל-MCP Server שג׳ירה עצמה מפעילה. להעביר את הטוקן של ג׳ירה אליה ואז לקבל את הפעולות שהוא יכול לבצע ומכאן לעבוד כמו MCP מקומי.
בעבר הרחוק (כלומר כמה חודשים לפני שכתבתי את הפוסט), הדרך היחידה היתה לעשות את זה עם SSE. אבל לפני כחודשיים נכנסה דרך חדשה שמוציאה לגמלאות את SSE והיא נקראת Streamable HTTP Transport. מאד פשוט ואינטואיטיבי להשתמש בה והיא משתמשת גם ב-SSE מאחורי הקלעים. נתחיל בהסבר על ה-server, נמשיך בקליינט ונסיים בהסבר מעמיק איך הstreamable מאחורי הקלעים.
הנה סרטון שהכנתי 🙂
הקוד בשרת
איך זה נראה ב-MCP Server? בגדול, די דומה. הקוד הוא פייתוני אבל להמיר אותו לג׳אווהסקריפט הוא כלום בפיתה. אני מדגים עם שרת ה-MCP העולב שלי, שיש לו רק כלי אחד שמבצע בדיקה של 5 הפוסטים האחרונים של האתר הזה. מה שמעניין אותנו זה לא פונקצית ה-fetch_rss שרק מביאה את ה-RSS וגם לא הפונקציה get_feed שהיא בעצם ״הכלי״ שה-MCP משתמש בו והוא זה שקורא ל-fetch_rss. מה שמעניין אותנו הוא החלק התחתון ביותר. הדגשתי אותו בקוד ואסביר עליו מייד אחרי הקוד:
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 with streamable-http transport
if __name__ == "__main__":
# Set server configuration through the settings property
mcp.settings.mount_path = "/mcp"
mcp.settings.port = 8765
mcp.settings.host = "127.0.0.1"
# Run the server with streamable-http transport
print(f"Starting MCP server at http://{mcp.settings.host}:{mcp.settings.port}{mcp.settings.mount_path}")
mcp.run(transport="streamable-http")מה שמעניין הוא שאני מגדיר את ה-mount_path של השרת, מגדיר את הפורט, במקרה הזה סתם פורט שבחרתי כמו 8765 ואת ההוסט, במקרה הזה לוקלהוסט. והשלב הכי חשוב הוא להגדיר ששרת ה-MCP יעבוד עם הטרנספורט של streamable-http.
מה, זה הכל?
כן! זה הכל. מבחינת שרת, להמיר משהו שעובד ב-stdio לשרת בסיסי שיכול לקבל בקשות מרחוק זה… רק לשנות את הטרנספורט!
הקוד בקליינט
מבחינת הקליינט, גם זה קל למדי. אם אתם משתמשים, אז כמובן שהקליינט הוא שקוף עבורכם. אתם יכולים להשתמש ב-mcp cli (שנכון לזמן כתיבת שורות אלו לא תומך ב-streamable אבל היוצר שלו אמר שזה עניין של ימים) או בקלאוד אנטרופיק. אני מצרף את הקוד הארוך והמשמים, אם אתם מממשים קליינט, אתם יכולים להדביק את זה ב-cursor או בקופיילוט ולבקש ממנו את אותו הדבר. אפשר גם לקחת קוד יותר פשוט מהדוקומנטציה אבל זה קוד שעובד לי ואני משתמש בו בדוגמה. אני אסביר על החלקים הרלוונטיים בקוד אחרי שאדביק:
import os
import asyncio
import json
from mcp.client.streamable_http import streamablehttp_client
from mcp import ClientSession
from openai import OpenAI
class SmartMCPClient:
"""An intelligent client that uses OpenAI to decide when to use MCP tools."""
def __init__(self, mcp_server_url=None, api_key=None):
# MCP server settings
self.mcp_server_url = mcp_server_url or "http://127.0.0.1:8765/mcp"
# OpenAI client for tool selection logic
self.api_key = api_key or os.environ.get("OPENAI_API_KEY")
if not self.api_key:
print("WARNING: No OpenAI API key provided. Set OPENAI_API_KEY environment variable.")
self.openai = OpenAI(api_key=self.api_key)
async def decide_tool_usage(self, query, available_tools):
"""
Use OpenAI to decide which tool to use based on the query
Args:
query: The user's question
available_tools: List of available tools from the MCP server
Returns:
tuple: (tool_name, tool_args) or (None, None) if no tool should be used
"""
# Quick filter for common greetings and chitchat - never use tools for these
common_greetings = ["hello", "hi", "hey", "greetings", "good morning", "good afternoon",
"how are you", "what's up", "howdy"]
# If query is just a simple greeting, don't use a tool
if query.lower() in common_greetings or len(query.split()) <= 2:
print("Query appears to be a simple greeting or too short - not using tools")
return None, None
if not self.api_key:
# More selective rule-based logic if no API key
relevant_terms = ['blog', 'post', 'article', 'feed', 'content', 'internet-israel']
# Only use the tool if query has specific relevant keywords
if 'get_feed' in available_tools and any(term in query.lower() for term in relevant_terms):
return 'get_feed', {'query': query}
return None, None
# Use OpenAI to decide
try:
tool_descriptions = {
'get_feed': "Gets the latest blog posts from Internet Israel. Use this for queries about blog posts, articles or content from internet-israel.com"
}
# Prepare descriptions for available tools
available_descriptions = {tool: tool_descriptions.get(tool, f"Tool: {tool}")
for tool in available_tools}
# Prepare the OpenAI prompt
messages = [
{"role": "system", "content": f"""You are a tool selection assistant.
Based on the user's query, determine if any of these available tools should be used:
{json.dumps(available_descriptions, indent=2)}
IMPORTANT: Only use a tool if the query is SPECIFICALLY asking about:
1. Blog posts or articles from Internet Israel
2. Content from internet-israel.com
3. Explicitly mentions blogs, posts, feeds, or articles
For general greetings, chitchat, or questions unrelated to Internet Israel content,
DO NOT use any tools. Examples where NO tool should be used:
- "Hello"
- "How are you?"
- "What's the weather like?"
- "Tell me about Python"
- "What time is it?"
If a tool should be used, respond in JSON format:
{{
"tool": "tool_name",
"args": {{
"query": "user query or relevant part"
}}
}}
If no tool should be used and you should answer the query directly, respond with:
{{
"tool": null,
"args": null
}}
Be selective and conservative with tool usage. Be concise. Only output valid JSON."""},
{"role": "user", "content": f"Query: {query}"}
]
# Call OpenAI
response = self.openai.chat.completions.create(
model="gpt-3.5-turbo",
messages=messages,
temperature=0.1,
response_format={"type": "json_object"}
)
# Parse the response
decision = json.loads(response.choices[0].message.content)
selected_tool = decision.get("tool")
args = decision.get("args", {})
if selected_tool and selected_tool in available_tools:
print(f"OpenAI decided to use tool: {selected_tool}")
return selected_tool, args
else:
print("OpenAI decided not to use any tools")
return None, None
except Exception as e:
print(f"Error using OpenAI for tool decision: {e}")
# Fall back to direct tool usage if error occurs
if 'get_feed' in available_tools:
return 'get_feed', {'query': query}
return None, None
async def process_query(self, query: str) -> str:
"""
Process a user query using a combination of OpenAI and MCP server.
Args:
query: The user's question or request
Returns:
The response as a string
"""
try:
print(f"Connecting to MCP server at {self.mcp_server_url}...")
async with streamablehttp_client(self.mcp_server_url) as (read_stream, write_stream, _):
print("Established streamable_http connection")
async with ClientSession(read_stream, write_stream) as session:
print("Created MCP client session")
# Initialize the session
await session.initialize()
print("Successfully initialized MCP session")
# Get available tools
try:
tools_response = await session.list_tools()
available_tools = [tool.name for tool in tools_response.tools]
print(f"Available tools: {available_tools}")
except Exception as e:
print(f"Could not list tools: {e}")
available_tools = []
# Use OpenAI to decide if a tool should be used
selected_tool, tool_args = await self.decide_tool_usage(query, available_tools)
# If OpenAI decided to use a tool, call it
if selected_tool:
print(f"Using tool: {selected_tool} with args: {tool_args}")
response = await session.call_tool(selected_tool, tool_args)
tool_result = response.content
print(f"Tool response received ({len(tool_result)} characters)")
# Use OpenAI to format the tool response nicely
if self.api_key:
try:
format_messages = [
{"role": "system", "content": f"You are a helpful assistant. Format and summarize the following raw data from the {selected_tool} tool in a user-friendly way:"},
{"role": "user", "content": tool_result}
]
format_response = self.openai.chat.completions.create(
model="gpt-3.5-turbo",
messages=format_messages,
temperature=0.7
)
formatted_result = format_response.choices[0].message.content
print("Formatted tool results with OpenAI")
return f"Results from Internet Israel blog:\n\n{formatted_result}"
except Exception as e:
print(f"Error formatting results: {e}")
return f"Tool results: {tool_result}"
else:
return f"Tool results: {tool_result}"
else:
# No suitable tool, use OpenAI to answer directly
if self.api_key:
try:
print(f"Using OpenAI to answer: '{query}'")
answer_messages = [
{"role": "system", "content": "You are a helpful assistant. Answer the user's question:"},
{"role": "user", "content": query}
]
answer_response = self.openai.chat.completions.create(
model="gpt-4", # Use GPT-4 for better responses
messages=answer_messages,
temperature=0.7
)
answer = answer_response.choices[0].message.content
return answer
except Exception as e:
print(f"Error using OpenAI: {e}")
return "Sorry, I couldn't process your request with OpenAI or MCP tools."
else:
return "No suitable tool found and OpenAI API key not provided."
except Exception as e:
import traceback
traceback.print_exc()
return f"Error: {str(e)}"
async def chat_loop(self):
"""Run an interactive chat loop."""
print("\nSmart MCP Client Started!")
print(f"Connected to: {self.mcp_server_url}")
print("Using OpenAI for intelligent tool selection: " + ("Yes" if self.api_key else "No (API key missing)"))
print("Type your queries or 'quit' to exit.")
while True:
try:
query = input("\nQuery: ").strip()
if query.lower() == 'quit':
break
print("\nProcessing...")
result = await self.process_query(query)
print("\nResponse:")
print(result)
except Exception as e:
print(f"\nError: {str(e)}")
async def main():
# Get API key from environment
api_key = os.environ.get("OPENAI_API_KEY")
if not api_key:
print("WARNING: No OPENAI_API_KEY found in environment variables.")
print("The client will work with limited capabilities.")
client = SmartMCPClient("http://127.0.0.1:8765/mcp", api_key)
await client.chat_loop()
if __name__ == "__main__":
asyncio.run(main())
מה שרלוונטי מבחינתנו הוא כמה שורות בודדות. בקליינט אנחנו צריכים להשתמש ב-streamable client:
from mcp.client.streamable_http import streamablehttp_client
ולהעביר אליו את ה-MCP Server באופן הזה – פה ממש נעשה החיבור.
streamablehttp_client(self.mcp_server_url) as (read_stream, write_stream, _):אחרי החיבור המוצלח, אני יכול ליצור session שיש לו session ID:
async with ClientSession(read_stream, write_stream) as session:
await session.initialize()יש session? אפשר לעבוד! זה החיבור האמיתי עם ה-MCP Server ואז אפשר לעבוד מולו כמו למשל לקבל את רשימת ה-tools:
tools_response = await session.list_tools()ומפה? בדיוק כמו בקליינט רגיל, אני אצטרך כן לתווך בין ה-LLM ל-MCP Server – לקבל את הפעולות, לשלוח אותן ל-LLM, לקבל מה מפעילים ואז לשלוח חזרה ל-MCP. על זה דיברנו כבר במאמר הקודם על קליינט.
מתחת למכסה המנוע – איך זה נראה?
אפשר לראות שזה די פשוט, אבל מה שמעניין הוא להבין כמה MCP, גם Streamable HTTP Transport, הוא לא קסם. בואו וננסה להבין איך הוא עובד מאחורי הקלעים. אני אשתמש בתוכנת Wireshark שמבצעת ניתוח של הרשת כדי לראות את המידע שעובר מ/אל השרת ולהבין מי מתעסק ומי מתרסק.
התקן ש-Streamable HTTP Transport עובדת איתו הוא תקן שנקרא JSON-RPC. ה-RPC הוא ראשי תבות של remote procedure call. לפני שאתם נרדמים ובורחים, זה שם מאד מסובך למשהו פשוט – דרך לקרוא לפונקציות (procedure) מרחוק. כשאני אומר דרך אני מתכוון לסטנדט. הרי אפשר לקרוא מרחוק לשרת דרך (למשל) בקשת GET. למשל https://example.com?activate_method=get_rss_feed ואז השרת יכול להבין שאני צריך לקרוא לפונקציה שיש לו get_rss_feed, להפעיל אותה ולהחזיר את המידע. הבעיה היא שמחר בבוקר מתכנת אחר יבוא וירצה לקרוא לפונקציה עם https://example.com?activate_function=get_rss_feed כי בחברה שלו השתמשו ב-activate_function ולא ב-activate_method. מה עושים? עושים סטנדרט וזה בדיוק מה ש-JSON-RPC עושה: תקן, דרך קבועה לעבוד. שעובדת עם JSON.
הדרך קובעת שאני מעביר לשרת אובייקט JSON עם:
- גרסת התקן
- המתודה שאני רוצה להפעיל.
- הפרמטרים
- מספר הזהות של הסשן.
למשל, משהו כזה, לשרת שעושה ״חיסור״ של מספר אחד לשני. 43 פחות 23.
{
"jsonrpc": "2.0",
"method": "subtract",
"params": [42, 23],
"id": 1
}
ואני אקבל חזרה תשובה שגם היא מוגדרת בתקן שהיא אובייקט JSON עם:
- גרסת התקן.
- התוצאה.
- מספר הזהות של הסשן.
{
"jsonrpc": "2.0",
"result": 19,
"id": 1
}
זה הכל? כן! כי ברגע שיש לנו תקן, אני לא צריך להסתבך בהסברים איך להתחבר לשירות שלי, אני פשוט אומר למתכנת שרוצה להתחבר שאני עובד עם JSON-RPC והוא אמור לדעת כבר איך להתחבר. גם אם הוא מגיע מחברה אחרת. זה הכוח של סטנדרט.
בחיבור הראשוני, ה-MCP קליינט שולח בקשת חיבור ל-MCP Server שנראית כך:
{
"method": "initialize",
"params": {
"protocolVersion": "2024-11-05",
"capabilities": {
"sampling": {},
"roots": {
"listChanged": true
}
},
"clientInfo": {
"name": "mcp",
"version": "0.1.0"
}
},
"jsonrpc": "2.0",
"id": 0
}אם הכל תקין והולידציה עוברת, חוזר 200 תקין עם ה-header של ה-MCP Session.
mcp-session-id: 5088b9fe7f0643beb546194841fbcca4
ועוד משהו חשוב: content-type: text/event-stream גם זה חלק מה-Header. השרת פה פותח תקשורת בפרוטקול SSE. עכשיו הוא יכול לשלוח לנו מידע על אותו חיבור של HTTP.
הקליינט שולח בקשה נוספת, הפעם עם notifications/initialized וה-mcp session. הכל תקין? מקבלים 200.
שלב לחיצת היד הושלם. השלב הבא הוא לקבל רשימת כלים. הקליינט שולח את זה:
{
"method": "tools/list",
"jsonrpc": "2.0",
"id": 1
}השרת עונה לו עם רשימת הכלים. פה זה שרת מסכן עם כלי אחד, די עלוב, אבל זה מה שיש – אבל שימו לב למה שמקבלים – התיאור, השם והארגומנטים מגיעים ממה שכתבנו בשרת.
{
"jsonrpc": "2.0",
"id": 1,
"result": {
"tools": [
{
"name": "get_feed",
"description": "Fetch latest blog posts from internet-israel.com\n\nArgs:\n limit: Number of feed items to return (default 5)\n",
"inputSchema": {
"properties": {
"limit": {
"default": 5,
"title": "Limit",
"type": "integer"
}
},
"title": "get_feedArguments",
"type": "object"
}
}
]
}
}אם הקליינט רוצה (במקרה הזה אם ה-LLM מנחה אותו), הוא שולח את ההפעלה לכלי באמצעות:
{
"method": "tools/call",
"params": {
"name": "get_feed",
"arguments": {
"query": "latest posts for internet-israel"
}
},
"jsonrpc": "2.0",
"id": 2
}וכמובן שהשרת יענה.
מה שחשוב לשים לב אליו זה שבסוף הסשן, הקליינט ישלח בקשת DELETE שיש בה את ה-mcp id שלנו כדי לאותת לשרת שאפשר לסגור את האופרציה הזו:
DELETE /mcp/ HTTP/1.1
Host: 127.0.0.1:8765
Accept-Encoding: gzip, deflate
Connection: keep-alive
User-Agent: python-httpx/0.28.1
Accept: application/json, text/event-stream
content-type: application/json
mcp-session-id: 2a8a30ff93f844f981eeee5b293b872eנכנסו קצת עמוק אל תוך ה-MCP. אבל הכניסה הזו היתה בעצם לראות שמבעד לכל הנצנצים והוודו של ה-MCP יש תקשורת שכולנו מכירים – שרת וקליינט, שמחליפים מידע ביניהם בפרוטוקול מוסכם שנקרא JSON-RPC שקובע פשוט את המבנה של ה-JSON שנשלח בין שני הצדדים. – יש תהליך לחיצת יד שבו מוחלף mcp-session-id. ואז הקליינט יכול לתקשר עם השרת באמצעות הפקודות המוגדרות בפרוטוקול MCP – כל עוד בנויות לפי פורמט JSON-RPC. זה הכל!
כמובן שכדאי להכיר את הפרוטוקול לעומק, כי ברגע שמתחילים לממש, אם לא מכירים קל ליפול לבעיות שונות ומשונות. אני למשל אכלתי קש וגבבה כי הקליינט שלי פנה ל-MCP באופן לא טוב, והשרת הגיב בהמון רידיירקטים – ובסקייל גדול היתה לי בעיה. כיוון שהכרתי ולא פחדתי לפתוח wireshark ולבדוק, הבנתי את הבעיה. זו העבודה הקשה של מתכנתים בעידן הנוכחי. לבנות ולפתח שירותים שמשתמשים ב-LLM, ולעשות אותם טובים. ובשביל זה צריך לא לפחד. אם קראתם עד לפה – הדרך שלכם סלולה 🙂
2 תגובות
רן, אתה יודע אולי אם יש כבר פתרונות לחיבור שרת MCP ל sharepoint של מייקרוסופט? רוב המידע הארגוני שלנו נמצא שם וחשבתי לבנות פתרון שיחבר את המידע הזה לאחד ממנועי השפה.
תודה.
אני יודע שיש
https://mcpmarket.com/server/sharepoint