▌Questions
》Question 1. Function description
get_weather_tool = {
"type": "function",
"name": "get_weather",
"description": "Get weather temperature for a specified city",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "The name of the city to get weather for"
}
},
"required": ["city"],
"additionalProperties": False
}
}
》Question 2. Another tool description
add another tool - a function that can add weather data to our database:
# 首先定義新函數（需要用到之前的 known_weather_data）
def set_weather(city: str, temp: float) -> None:
city = city.strip().lower()
known_weather_data[city] = temp
return 'OK'
# 然後定義工具描述
set_weather_tool = {
"type": "function",
"name": "set_weather",
"description": "Set weather temperature for a specified city",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "The name of the city to set weather for"
},
"temp": {
"type": "number",
"description": "The temperature value to set for the city"
}
},
"required": ["city", "temp"],
"additionalProperties": False
}
}
》Question 3. FastMCP version (1 point)
Name: fastmcp
Version: 2.10.5
Summary: The fast, Pythonic way to build MCP servers and clients.
Home-page: https://gofastmcp.com
Author: Jeremiah Lowin
》Question 4. MCP Server transport (1 point)
# weather_server.py
import random
from fastmcp import FastMCP
mcp = FastMCP("Demo 🚀")
known_weather_data = {
'berlin': 20.0
}
@mcp.tool
def get_weather(city: str) -> float:
"""
Retrieves the temperature for a specified city.
Parameters:
city (str): The name of the city for which to retrieve weather data.
Returns:
float: The temperature associated with the city.
"""
city = city.strip().lower()
if city in known_weather_data:
return known_weather_data[city]
return round(random.uniform(-5, 35), 1)
@mcp.tool
def set_weather(city: str, temp: float) -> None:
"""
Sets the temperature for a specified city.
Parameters:
city (str): The name of the city for which to set the weather data.
temp (float): The temperature to associate with the city.
Returns:
str: A confirmation string 'OK' indicating successful update.
"""
city = city.strip().lower()
known_weather_data[city] = temp
return 'OK'
if __name__ == "__main__":
mcp.run()
[07/15/25 21:45:01] INFO Starting MCP server 'Demo 🚀' with transport 'stdio'
》Question 5. MCP communication (1 point)
{"jsonrpc": "2.0", "id": 3, "method": "tools/call", "params": {"name": "get_weather", "arguments": {"city": "Berlin"}}}
{"jsonrpc":"2.0","id":3,"result":{"content":[{"type":"text","text":"20.0"}],"structuredContent":{"result":20.0},"isError":false}}
》Question 6. MCP Client tools (1 point)
[
{
"name": "get_weather",
"description": "Retrieves the temperature for a specified city.\n\nParameters:\n city (str): The name of the city for which to retrieve weather data.\n\nReturns:\n float: The temperature associated with the city.",
"inputSchema": {
"properties": {
"city": {
"title": "City",
"type": "string"
}
},
"required": ["city"],
"type": "object"
},
"outputSchema": {
"properties": {
"result": {
"title": "Result",
"type": "number"
}
},
"required": ["result"],
"title": "_WrappedResult",
"type": "object",
"x-fastmcp-wrap-result": true
}
},
{
"name": "set_weather",
"description": "Sets the temperature for a specified city.\n\nParameters:\n city (str): The name of the city for which to set the weather data.\n temp (float): The temperature to associate with the city.\n\nReturns:\n str: A confirmation string 'OK' indicating successful update.",
"inputSchema": {
"properties": {
"city": {
"title": "City",
"type": "string"
},
"temp": {
"title": "Temp",
"type": "number"
}
},
"required": ["city", "temp"],
"type": "object"
}
}
]
▌weather_server.py
# weather_server.py
import random
from fastmcp import FastMCP
mcp = FastMCP("Demo 🚀")
known_weather_data = {
'berlin': 20.0
}
@mcp.tool
def get_weather(city: str) -> float:
"""
Retrieves the temperature for a specified city.
Parameters:
city (str): The name of the city for which to retrieve weather data.
Returns:
float: The temperature associated with the city.
"""
city = city.strip().lower()
if city in known_weather_data:
return known_weather_data[city]
return round(random.uniform(-5, 35), 1)
@mcp.tool
def set_weather(city: str, temp: float) -> None:
"""
Sets the temperature for a specified city.
Parameters:
city (str): The name of the city for which to set the weather data.
temp (float): The temperature to associate with the city.
Returns:
str: A confirmation string 'OK' indicating successful update.
"""
city = city.strip().lower()
known_weather_data[city] = temp
return 'OK'
if __name__ == "__main__":
mcp.run()
▌mcp_client.py
import json
import subprocess
from typing import Dict, Any, List, Optional
class MCPClient:
def __init__(self, server_command: List[str]):
"""
Initialize the FastMCP client.
Args:
server_command: Command to start the server (e.g., ["python", "server.py"])
"""
self.server_command = server_command
self.process = None
self.request_id = 0
self.available_tools = {}
self.is_initialized = False
def start_server(self):
"""Start the FastMCP server process"""
self.process = subprocess.Popen(
self.server_command,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
bufsize=0,
encoding='utf-8'
)
print(f"Started server with command: {' '.join(self.server_command)}")
def stop_server(self):
"""Stop the server process"""
if self.process:
self.process.terminate()
self.process.wait()
print("Server stopped")
def _get_next_request_id(self) -> int:
"""Get the next request ID"""
self.request_id += 1
return self.request_id
def _send_notification(self, method: str, params: Optional[Dict[str, Any]] = None):
"""Send a notification (no response expected)"""
if not self.process:
raise RuntimeError("Server not started")
notification = {
"jsonrpc": "2.0",
"method": method
}
if params:
notification["params"] = params
# Send notification
notification_str = json.dumps(notification) + "\n"
self.process.stdin.write(notification_str)
self.process.stdin.flush()
def _send_request(self, method: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
"""Send a JSON-RPC request to the server via stdin"""
if not self.process:
raise RuntimeError("Server not started")
request = {
"jsonrpc": "2.0",
"id": self._get_next_request_id(),
"method": method
}
if params:
request["params"] = params
# Send request
request_str = json.dumps(request) + "\n"
self.process.stdin.write(request_str)
self.process.stdin.flush()
# Read response
response_str = self.process.stdout.readline().strip()
if not response_str:
raise RuntimeError("No response from server")
response = json.loads(response_str)
if "error" in response:
raise Exception(f"Server error: {response['error']}")
return response.get("result", {})
def initialize(self) -> Dict[str, Any]:
"""Send initialize request to the server"""
print("Sending initialize request...")
result = self._send_request(
"initialize",
{
"protocolVersion": "2024-11-05",
"capabilities": {
"roots": {"listChanged": True},
"sampling": {}
},
"clientInfo": {
"name": "test-client",
"version": "1.0.0"
}
}
)
print(f"Initialize response: {result}")
return result
def initialized(self):
"""Send initialized notification to complete handshake"""
print("Sending initialized notification...")
self._send_notification("notifications/initialized")
self.is_initialized = True
print("Handshake completed successfully")
def get_tools(self) -> List[Dict[str, Any]]:
"""Get available tools from the server"""
if not self.is_initialized:
raise RuntimeError("Client not initialized. Call initialize() and initialized() first.")
print("Retrieving available tools...")
result = self._send_request("tools/list")
tools = result.get("tools", [])
# Store tools for easy access
self.available_tools = {tool["name"]: tool for tool in tools}
print(f"Available tools: {list(self.available_tools.keys())}")
return tools
def call_tool(self, tool_name: str, arguments: Dict[str, Any]) -> Any:
"""Call a specific tool with given arguments"""
if not self.is_initialized:
raise RuntimeError("Client not initialized. Call initialize() and initialized() first.")
if tool_name not in self.available_tools:
raise ValueError(f"Tool '{tool_name}' not available. Available tools: {list(self.available_tools.keys())}")
print(f"Calling tool '{tool_name}' with arguments: {arguments}")
result = self._send_request(
"tools/call",
{
"name": tool_name,
"arguments": arguments
}
)
return result
def list_available_tools(self):
"""Print information about available tools"""
if not self.available_tools:
print("No tools available. Call get_tools() first.")
return
print("\nAvailable Tools:")
print("-" * 50)
for name, tool in self.available_tools.items():
print(f"Name: {name}")
print(f"Description: {tool.get('description', 'No description')}")
# Print input schema if available
input_schema = tool.get('inputSchema', {})
if input_schema.get('properties'):
print("Parameters:")
for param_name, param_info in input_schema['properties'].items():
param_type = param_info.get('type', 'unknown')
param_desc = param_info.get('description', 'No description')
print(f" - {param_name} ({param_type}): {param_desc}")
print("-" * 50)
def convert_mcp_tool_to_function_format(mcp_tool):
"""
Convert MCP tool format to function format.
Args:
mcp_tool: Tool object or dict with MCP format
Returns:
dict: Tool in function format
"""
# Handle both Tool objects and dictionaries
if hasattr(mcp_tool, 'name'):
# It's a Tool object
name = mcp_tool.name
description = mcp_tool.description
input_schema = mcp_tool.inputSchema
else:
# It's a dictionary
name = mcp_tool['name']
description = mcp_tool['description']
input_schema = mcp_tool['inputSchema']
# Clean up description - remove docstring formatting
clean_description = description.split('\n\n')[0] if '\n\n' in description else description
clean_description = clean_description.strip()
# Convert the tool format
function_tool = {
"type": "function",
"name": name,
"description": clean_description,
"parameters": {
"type": "object",
"properties": {},
"required": input_schema.get('required', []),
"additionalProperties": False
}
}
# Convert properties
if 'properties' in input_schema:
for prop_name, prop_info in input_schema['properties'].items():
function_tool["parameters"]["properties"][prop_name] = {
"type": prop_info.get('type', 'string'),
"description": prop_info.get('description', f"{prop_name.replace('_', ' ').title()}")
}
# Add title as description if no description exists
if 'title' in prop_info and 'description' not in prop_info:
function_tool["parameters"]["properties"][prop_name]["description"] = prop_info['title']
return function_tool
def convert_tools_list(mcp_tools):
"""
Convert a list of MCP tools to function format.
Args:
mcp_tools: List of MCP tools
Returns:
list: List of tools in function format
"""
return [convert_mcp_tool_to_function_format(tool) for tool in mcp_tools]
class MCPTools:
def __init__(self, mcp_client):
self.mcp_client = mcp_client
self.tools = None
def get_tools(self):
if self.tools is None:
mcp_tools = self.mcp_client.get_tools()
self.tools = convert_tools_list(mcp_tools)
return self.tools
def function_call(self, tool_call_response):
function_name = tool_call_response.name
arguments = json.loads(tool_call_response.arguments)
result = self.mcp_client.call_tool(function_name, arguments)
return {
"type": "function_call_output",
"call_id": tool_call_response.call_id,
"output": json.dumps(result, indent=2),
}
▌async_client.py
import asyncio
from fastmcp import Client
async def main():
async with Client("weather_server.py") as mcp_client:
tools = await mcp_client.list_tools()
return tools
if __name__ == "__main__":
result = asyncio.run(main())
print(result)
▌test_client.py
# test_client.py
import mcp_client
def main():
# 建立客戶端實例
our_mcp_client = mcp_client.MCPClient(["python", "weather_server.py"])
try:
# 啟動服務器連接
our_mcp_client.start_server()
# 初始化連接
our_mcp_client.initialize()
# 完成初始化
our_mcp_client.initialized()
# 獲取工具列表（Q6 答案）
print("=== Q6 答案：可用工具列表 ===")
tools = our_mcp_client.get_tools()
print(tools)
# 測試工具調用
print("\n=== 測試工具調用 ===")
result1 = our_mcp_client.call_tool('get_weather', {'city': 'Berlin'})
print(f"Berlin 天氣：{result1}")
finally:
# 清理
our_mcp_client.stop_server()
if __name__ == "__main__":
main()