関連記事
- MCPサーバーをPythonで構築(1) 基礎
- MCPサーバーをPythonで構築(2) サーバー構築 ← このページ
実装する仕様
早速PythonでMCPサーバーを構築していきます。
以下の仕様とします。
・県名を受け取り、対応する観光地を1つ返す
・最小構成での実装
・ポート8000でlocalhostで起動
・県名と観光地のマッピングデータ観光地のデータはプログラム内で直接記載
・簡単なエラーハンドリングも実装
・最小構成での実装
・ポート8000でlocalhostで起動
・県名と観光地のマッピングデータ観光地のデータはプログラム内で直接記載
・簡単なエラーハンドリングも実装
※実際にはプロンプトを受け取り結果を返すことが多いが、今回は簡易的に県名を受け取ることとする。
プログラムファイルのフォルダ構成
なるべく最小限として、以下の構成とします。
mcp-server/ # メインプロジェクトフォルダ
error_handler.py # エラーハンドリング機能
json_rpc_handler.py # JSON-RPC処理機能
main.py # メイン処理
mcp_server_sdk.py # SDK版MCPサーバー
tourism_data.py # 観光地データ定義
requirements.txt # 基本依存関係
実装したプログラム
今回実装したプログラムは、AIを使用して開発しています。
詳しくはこちらの開発手法を参照ください。
requirements.txt
Python プロジェクトで使う パッケージ（ライブラリ）の依存関係を管理するファイル です。
今回はmcpを使用するため以下のように定義します。
mcp>=1.16.0
tourism_data.py
都道府県と観光地のマッピングデータです。
通常はベクトルDB等を使用しますが、今回は直接ファイルに記載します。
# -*- coding: utf-8 -*-
"""
観光地データ管理モジュール
47都道府県の観光地データを管理する。
"""
# 47都道府県の観光地データ（観光地名 + 簡単な説明）
TOURISM_DATA = {
"北海道": "函館山 - 夜景が美しく、海と街の光景が広がる山です。",
"青森県": "奥入瀬渓流 - 四季折々の自然が楽しめる清流です。",
"岩手県": "中尊寺 - 金色堂が有名な歴史ある寺院です。",
"宮城県": "松島 - 日本三景のひとつ、美しい島々が点在します。",
"秋田県": "角館武家屋敷 - 江戸時代の街並みが残る武家屋敷通りです。",
"山形県": "山寺 - 崖に建つ古刹で、登山と景観が楽しめます。",
"福島県": "会津若松城 - 鶴ヶ城とも呼ばれる美しい城です。",
"茨城県": "偕楽園 - 梅の名所として有名な日本庭園です。",
"栃木県": "日光東照宮 - 豪華絢爛な装飾が特徴の世界遺産です。",
"群馬県": "草津温泉 - 湯畑が名物の日本有数の温泉地です。",
"埼玉県": "川越 - 小江戸情緒あふれる町並みが楽しめます。",
"千葉県": "成田山新勝寺 - 厄除けで有名な歴史ある寺院です。",
"東京都": "浅草寺 - 雷門や仲見世通りで賑わう東京の名所です。",
"神奈川県": "鎌倉大仏 - 高徳院にある巨大な青銅の仏像です。",
"新潟県": "佐渡島 - 自然と歴史が融合した魅力的な島です。",
"富山県": "立山黒部アルペンルート - 壮大な雪の壁と山岳景観です。",
"石川県": "兼六園 - 四季の移ろいを楽しめる名園です。",
"福井県": "東尋坊 - 絶景の断崖で、荒波の景観が楽しめます。",
"山梨県": "富士山 - 日本の象徴であり、登山や景観が楽しめます。",
"長野県": "善光寺 - 歴史深いお寺で参拝と文化体験が可能です。",
"岐阜県": "白川郷 - 合掌造りの家屋が並ぶ世界遺産です。",
"静岡県": "富士山 - 世界文化遺産に登録された美しい山です。",
"愛知県": "名古屋城 - 金の鯱が目印の歴史ある城です。",
"三重県": "伊勢神宮 - 日本屈指の神社で参拝の名所です。",
"滋賀県": "琵琶湖 - 湖畔で遊覧や自然散策が楽しめます。",
"京都府": "清水寺 - 眺望が美しく、歴史と文化の名所です。",
"大阪府": "大阪城 - 豊臣秀吉ゆかりの大きな城です。",
"兵庫県": "姫路城 - 白鷺城と呼ばれる美しい世界遺産です。",
"奈良県": "東大寺 - 大仏殿で知られる歴史ある寺院です。",
"和歌山県": "高野山 - 世界遺産で、修行と観光の名所です。",
"鳥取県": "鳥取砂丘 - 風と砂が織りなす、魅力あふれる砂丘です。",
"島根県": "出雲大社 - 縁結びで有名な古社です。",
"岡山県": "岡山後楽園 - 日本三名園のひとつ、美しい庭園です。",
"広島県": "厳島神社 - 海上に浮かぶ鳥居が印象的な世界遺産です。",
"山口県": "角島大橋 - 青い海に架かる絶景の橋です。",
"徳島県": "鳴門の渦潮 - 大迫力の渦潮が見られる観光名所です。",
"香川県": "金刀比羅宮 - 海の安全を祈る歴史ある神社です。",
"愛媛県": "道後温泉 - 日本最古の温泉地で風情があります。",
"高知県": "桂浜 - 太平洋を望む美しい海岸です。",
"福岡県": "太宰府天満宮 - 学問の神様を祀る人気の神社です。",
"佐賀県": "吉野ヶ里遺跡 - 弥生時代の歴史を感じる遺跡です。",
"長崎県": "ハウステンボス - オランダの街並みを再現したテーマパークです。",
"熊本県": "阿蘇山 - 活火山と大自然を楽しめる観光地です。",
"大分県": "別府温泉 - 湯けむり立ち上る温泉街が魅力です。",
"宮崎県": "高千穂峡 - 渓谷美と神話の里が楽しめます。",
"鹿児島県": "桜島 - 活火山を間近に感じられる観光地です。",
"沖縄県": "首里城 - 琉球王国の歴史を伝える城跡です。"
}
def get_tourist_spot(prefecture):
"""
県名から観光地を取得する
Args:
prefecture (str): 県名
Returns:
str: 観光地名。該当する県名がない場合はNoneを返す
"""
return TOURISM_DATA.get(prefecture)
def get_all_prefectures():
"""
全ての都道府県名のリストを取得する
Returns:
list: 都道府県名のリスト
"""
return list(TOURISM_DATA.keys())
def is_valid_prefecture(prefecture):
"""
指定された県名が有効かどうかを確認する
Args:
prefecture (str): 県名
Returns:
bool: 有効な県名の場合はTrue、そうでなければFalse
"""
return prefecture in TOURISM_DATA
def get_data_count():
"""
登録されている都道府県数を取得する
Returns:
int: 都道府県数
"""
return len(TOURISM_DATA)
main.py
起動するMCPサーバーです。
# -*- coding: utf-8 -*-
"""
MCP Tourism Server
県名を受け取り、対応する観光地を返すMCPサーバー
JSON-RPC 2.0プロトコルに準拠
"""
import json
import sys
from http.server import HTTPServer, BaseHTTPRequestHandler
from urllib.parse import urlparse
from json_rpc_handler import JSONRPCHandler
from error_handler import ErrorHandler
class MCPTourismServer:
"""MCP Tourism Server メインクラス"""
def __init__(self, host='localhost', port=8000):
"""
サーバーを初期化する
Args:
host (str): ホスト名
port (int): ポート番号
"""
self.host = host
self.port = port
self.json_rpc_handler = JSONRPCHandler()
self.error_handler = ErrorHandler()
def start_server(self):
"""サーバーを起動する"""
try:
server_address = (self.host, self.port)
httpd = HTTPServer(server_address, MCPRequestHandler)
httpd.json_rpc_handler = self.json_rpc_handler
httpd.error_handler = self.error_handler
print(f"MCP Tourism Server が起動しました")
print(f"ホスト: {self.host}")
print(f"ポート: {self.port}")
print(f"URL: http://{self.host}:{self.port}")
print("サーバーを停止するには Ctrl+C を押してください")
print("-" * 50)
httpd.serve_forever()
except KeyboardInterrupt:
print("\nサーバーを停止しています...")
httpd.shutdown()
print("サーバーが停止しました")
except Exception as e:
print(f"サーバー起動エラー: {e}")
sys.exit(1)
class MCPRequestHandler(BaseHTTPRequestHandler):
"""HTTPリクエストハンドラー"""
def do_POST(self):
"""POSTリクエストを処理する"""
try:
# Content-Lengthの確認
content_length = int(self.headers.get('Content-Length', 0))
if content_length == 0:
self._send_error_response(400, "Request body is empty")
return
# リクエストボディを一度だけ読み取る
request_body = self.rfile.read(content_length).decode('utf-8')
print("=== Received POST body ===")
print(request_body)
# Content-Typeの確認
content_type = self.headers.get('Content-Type', '')
if 'application/json' not in content_type:
self._send_error_response(400, "Content-Type must be application/json")
return
# JSON-RPCリクエストを解析（self.server.json_rpc_handler を使う）
request_data, error_response = self.server.json_rpc_handler.parse_request(request_body)
if error_response:
self._send_json_response(error_response)
return
# JSON-RPCリクエストを処理
response_data = self.server.json_rpc_handler.handle_request(request_data)
# レスポンス送信
self._send_json_response(response_data)
except Exception as e:
import traceback
print("Error in do_POST:", e)
traceback.print_exc()
self.send_response(500)
self.send_header('Content-Type', 'application/json')
self.end_headers()
self.wfile.write(b'Internal Server Error')
def do_GET(self):
"""GETリクエストを処理する（ヘルスチェック用）"""
if self.path == '/health':
self._send_health_response()
elif self.path == '/':
self._send_info_response()
else:
self._send_error_response(404, "Not Found")
def _send_json_response(self, data):
"""JSONレスポンスを送信する"""
try:
response_json = self.server.json_rpc_handler.create_response_json(data)
self.send_response(200)
self.send_header('Content-Type', 'application/json; charset=utf-8')
self.send_header('Content-Length', str(len(response_json.encode('utf-8'))))
self.end_headers()
self.wfile.write(response_json.encode('utf-8'))
except Exception as e:
self._send_error_response(500, f"Response generation error: {str(e)}")
def _send_health_response(self):
"""ヘルスチェックレスポンスを送信する"""
health_data = {
"status": "healthy",
"service": "MCP Tourism Server",
"version": "1.0.0"
}
self._send_json_response(health_data)
def _send_info_response(self):
"""サーバー情報レスポンスを送信する"""
info_data = {
"service": "MCP Tourism Server",
"version": "1.0.0",
"description": "県名を受け取り、対応する観光地を返すMCPサーバー",
"protocol": "JSON-RPC 2.0",
"methods": ["get_tourist_spot"],
"endpoints": {
"POST /": "JSON-RPC endpoint",
"GET /health": "Health check",
"GET /": "Service information"
}
}
self._send_json_response(info_data)
def _send_error_response(self, status_code, message):
"""エラーレスポンスを送信する"""
error_data = {
"error": {
"code": status_code,
"message": message
}
}
try:
response_json = json.dumps(error_data, ensure_ascii=False, indent=2)
self.send_response(status_code)
self.send_header('Content-Type', 'application/json; charset=utf-8')
self.send_header('Content-Length', str(len(response_json.encode('utf-8'))))
self.end_headers()
self.wfile.write(response_json.encode('utf-8'))
except Exception as e:
self.send_response(500)
self.end_headers()
self.wfile.write(b'Internal Server Error')
def log_message(self, format, *args):
"""ログメッセージを出力する"""
print(f"[{self.log_date_time_string()}] {format % args}")
def main():
"""メイン関数"""
import argparse
parser = argparse.ArgumentParser(description='MCP Tourism Server')
parser.add_argument('--host', default='localhost', help='Host name (default: localhost)')
parser.add_argument('--port', type=int, default=8000, help='Port number (default: 8000)')
args = parser.parse_args()
# サーバーを起動
server = MCPTourismServer(host=args.host, port=args.port)
server.start_server()
if __name__ == '__main__':
main()
mcp_server_sdk.py
MCP Python SDKを使用して、県名から対応する観光地を返します。
# -*- coding: utf-8 -*-
"""
MCP Tourism Server (SDK版)
公式MCP Python SDKを使用した県名を受け取り、対応する観光地を返すMCPサーバー
"""
import asyncio
import logging
from typing import Any, Sequence
from mcp.server import Server
from mcp.server.models import InitializationOptions
from mcp.server.stdio import stdio_server
from mcp.types import (
CallToolRequest,
CallToolResult,
ListToolsRequest,
ListToolsResult,
Tool,
TextContent,
)
from tourism_data import get_tourist_spot, is_valid_prefecture, get_all_prefectures
# ログ設定
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# MCPサーバーのインスタンスを作成
server = Server("tourism-server")
@server.list_tools()
async def handle_list_tools() -> ListToolsResult:
"""
利用可能なツールのリストを返す
"""
return ListToolsResult(
tools=[
Tool(
name="get_tourist_spot",
description="県名を指定して観光地を取得する",
inputSchema={
"type": "object",
"properties": {
"prefecture": {
"type": "string",
"description": "都道府県名（例: 東京都、北海道）",
"maxLength": 50
}
},
"required": ["prefecture"]
}
),
Tool(
name="list_prefectures",
description="利用可能な都道府県のリストを取得する",
inputSchema={
"type": "object",
"properties": {}
}
)
]
)
@server.call_tool()
async def handle_call_tool(name: str, arguments: dict[str, Any]) -> CallToolResult:
"""
ツール呼び出しを処理する
"""
try:
if name == "get_tourist_spot":
return await handle_get_tourist_spot(arguments)
elif name == "list_prefectures":
return await handle_list_prefectures(arguments)
else:
return CallToolResult(
content=[TextContent(
type="text",
text=f"未知のツール: {name}"
)],
isError=True
)
except Exception as e:
logger.error(f"ツール呼び出しエラー: {e}")
return CallToolResult(
content=[TextContent(
type="text",
text=f"エラーが発生しました: {str(e)}"
)],
isError=True
)
async def handle_get_tourist_spot(arguments: dict[str, Any]) -> CallToolResult:
"""
観光地取得ツールを処理する
"""
prefecture = arguments.get("prefecture", "").strip()
# 入力値検証
if not prefecture:
return CallToolResult(
content=[TextContent(
type="text",
text="エラー: 県名が指定されていません"
)],
isError=True
)
if len(prefecture) > 50:
return CallToolResult(
content=[TextContent(
type="text",
text="エラー: 県名が長すぎます（最大50文字）"
)],
isError=True
)
# 県名の存在確認
if not is_valid_prefecture(prefecture):
available_prefectures = ", ".join(get_all_prefectures()[:10]) # 最初の10個のみ表示
return CallToolResult(
content=[TextContent(
type="text",
text=f"エラー: 指定された県名 '{prefecture}' が見つかりません。\n利用可能な県名の例: {available_prefectures}..."
)],
isError=True
)
# 観光地を取得
tourist_spot = get_tourist_spot(prefecture)
return CallToolResult(
content=[TextContent(
type="text",
text=f"【{prefecture}】の観光地: {tourist_spot}"
)]
)
async def handle_list_prefectures(arguments: dict[str, Any]) -> CallToolResult:
"""
都道府県リスト取得ツールを処理する
"""
prefectures = get_all_prefectures()
# 都道府県を10個ずつに分けて表示
result_text = "【利用可能な都道府県】\n\n"
for i in range(0, len(prefectures), 10):
chunk = prefectures[i:i+10]
result_text += " ".join(chunk) + "\n"
result_text += f"\n合計: {len(prefectures)}都道府県"
return CallToolResult(
content=[TextContent(
type="text",
text=result_text
)]
)
async def main():
"""
メイン関数
"""
logger.info("MCP Tourism Server (SDK版) を起動しています...")
# サーバー情報をログ出力
logger.info("利用可能なツール:")
logger.info(" - get_tourist_spot: 県名を指定して観光地を取得")
logger.info(" - list_prefectures: 利用可能な都道府県のリストを取得")
# stdioサーバーを起動
async with stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
InitializationOptions(
server_name="tourism-server",
server_version="1.0.0",
capabilities=server.get_capabilities(
notification_options=None,
experimental_capabilities=None,
),
),
)
if __name__ == "__main__":
asyncio.run(main())
json_rpc_handler.py
JSON-RPCに準拠したリクエスト及び、レスポンス処理を行います。
# -*- coding: utf-8 -*-
"""
JSON-RPC処理モジュール
JSON-RPC 2.0プロトコルに準拠したリクエスト・レスポンス処理を行う。
"""
import json
from tourism_data import get_tourist_spot, is_valid_prefecture
from error_handler import ErrorHandler
class JSONRPCHandler:
"""JSON-RPC処理クラス"""
def __init__(self):
self.error_handler = ErrorHandler()
def handle_request(self, request_data):
"""
JSON-RPCリクエストを処理する
Args:
request_data (dict): リクエストデータ
Returns:
dict: レスポンスデータ
"""
try:
# リクエストの妥当性を検証
is_valid, error_response = self.error_handler.validate_json_rpc_request(request_data)
if not is_valid:
return error_response
# メソッド名を取得
method = request_data.get("method")
params = request_data.get("params", {})
request_id = request_data.get("id")
# メソッドの分岐処理
if method == "get_tourist_spot":
return self._handle_get_tourist_spot(params, request_id)
else:
return self.error_handler.create_method_not_found_error(method, request_id)
except Exception as e:
return self.error_handler.create_internal_error(
f"予期しないエラーが発生しました: {str(e)}",
request_data.get("id")
)
def _handle_get_tourist_spot(self, params, request_id):
"""
観光地取得メソッドを処理する
Args:
params (dict): パラメータ
request_id: リクエストID
Returns:
dict: レスポンスデータ
"""
# パラメータの妥当性を検証
is_valid, error_message = self.error_handler.validate_prefecture_param(params)
if not is_valid:
return self.error_handler.create_invalid_params_error(error_message, request_id)
prefecture = params.get("prefecture")
# 県名の存在確認
if not is_valid_prefecture(prefecture):
return self.error_handler.create_invalid_params_error(
f"指定された県名が見つかりません: {prefecture}",
request_id
)
# 観光地を取得
tourist_spot = get_tourist_spot(prefecture)
# 成功レスポンスを生成
return self._create_success_response(prefecture, tourist_spot, request_id)
def _create_success_response(self, prefecture, tourist_spot, request_id):
"""
成功レスポンスを生成する
Args:
prefecture (str): 県名
tourist_spot (str): 観光地名
request_id: リクエストID
Returns:
dict: 成功レスポンス
"""
return {
"jsonrpc": "2.0",
"result": {
"prefecture": prefecture,
"tourist_spot": tourist_spot
},
"id": request_id
}
def parse_request(self, request_body):
"""
リクエストボディを解析する
Args:
request_body (str): リクエストボディ（JSON文字列）
Returns:
tuple: (request_data, error_response)
- request_data (dict or None): 解析されたリクエストデータ
- error_response (dict or None): エラーレスポンス（解析失敗時）
"""
try:
request_data = json.loads(request_body)
return request_data, None
except json.JSONDecodeError as e:
error_response = self.error_handler.create_parse_error()
return None, error_response
except Exception as e:
error_response = self.error_handler.create_internal_error(
f"リクエスト解析エラー: {str(e)}"
)
return None, error_response
def create_response_json(self, response_data):
"""
レスポンスデータをJSON文字列に変換する
Args:
response_data (dict): レスポンスデータ
Returns:
str: JSON文字列
"""
try:
return json.dumps(response_data, ensure_ascii=False, indent=2)
except Exception as e:
# JSON変換エラーの場合は内部エラーレスポンスを返す
error_response = self.error_handler.create_internal_error(
f"レスポンス生成エラー: {str(e)}"
)
return json.dumps(error_response, ensure_ascii=False, indent=2)
error_handler.py
エラーハンドリングの処理です。
# -*- coding: utf-8 -*-
"""
エラーハンドリングモジュール
JSON-RPC標準に準拠したエラーレスポンスを生成する。
"""
import json
class MCPError(Exception):
"""MCPサーバー用のカスタム例外クラス"""
def __init__(self, code, message, data=None):
self.code = code
self.message = message
self.data = data
super().__init__(self.message)
class ErrorHandler:
"""エラーハンドリングクラス"""
# JSON-RPC 2.0 エラーコード定義
PARSE_ERROR = -32700
INVALID_REQUEST = -32600
METHOD_NOT_FOUND = -32601
INVALID_PARAMS = -32602
INTERNAL_ERROR = -32603
@staticmethod
def create_error_response(error_code, message, request_id=None, data=None):
"""
エラーレスポンスを生成する
Args:
error_code (int): エラーコード
message (str): エラーメッセージ
request_id: リクエストID（Noneの場合はnull）
data: 追加データ（オプション）
Returns:
dict: JSON-RPC形式のエラーレスポンス
"""
error_response = {
"jsonrpc": "2.0",
"error": {
"code": error_code,
"message": message
},
"id": request_id
}
if data is not None:
error_response["error"]["data"] = data
return error_response
@staticmethod
def create_parse_error(request_id=None):
"""
パースエラーレスポンスを生成する
Args:
request_id: リクエストID
Returns:
dict: パースエラーレスポンス
"""
return ErrorHandler.create_error_response(
ErrorHandler.PARSE_ERROR,
"Parse error",
request_id
)
@staticmethod
def create_invalid_request_error(request_id=None):
"""
無効リクエストエラーレスポンスを生成する
Args:
request_id: リクエストID
Returns:
dict: 無効リクエストエラーレスポンス
"""
return ErrorHandler.create_error_response(
ErrorHandler.INVALID_REQUEST,
"Invalid Request",
request_id
)
@staticmethod
def create_method_not_found_error(method_name, request_id=None):
"""
メソッド未発見エラーレスポンスを生成する
Args:
method_name (str): メソッド名
request_id: リクエストID
Returns:
dict: メソッド未発見エラーレスポンス
"""
return ErrorHandler.create_error_response(
ErrorHandler.METHOD_NOT_FOUND,
"Method not found",
request_id,
f"指定されたメソッドが見つかりません: {method_name}"
)
@staticmethod
def create_invalid_params_error(message, request_id=None):
"""
無効パラメータエラーレスポンスを生成する
Args:
message (str): エラーメッセージ
request_id: リクエストID
Returns:
dict: 無効パラメータエラーレスポンス
"""
return ErrorHandler.create_error_response(
ErrorHandler.INVALID_PARAMS,
"Invalid params",
request_id,
message
)
@staticmethod
def create_internal_error(message, request_id=None):
"""
内部サーバーエラーレスポンスを生成する
Args:
message (str): エラーメッセージ
request_id: リクエストID
Returns:
dict: 内部サーバーエラーレスポンス
"""
return ErrorHandler.create_error_response(
ErrorHandler.INTERNAL_ERROR,
"Internal error",
request_id,
message
)
@staticmethod
def validate_json_rpc_request(request_data):
"""
JSON-RPCリクエストの妥当性を検証する
Args:
request_data (dict): リクエストデータ
Returns:
tuple: (is_valid, error_response)
- is_valid (bool): 妥当性
- error_response (dict or None): エラーレスポンス（無効な場合）
"""
# 必須フィールドの確認
required_fields = ["jsonrpc", "method"]
for field in required_fields:
if field not in request_data:
return False, ErrorHandler.create_invalid_request_error(
request_data.get("id")
)
# jsonrpcバージョンの確認
if request_data.get("jsonrpc") != "2.0":
return False, ErrorHandler.create_invalid_request_error(
request_data.get("id")
)
# メソッド名の確認
method = request_data.get("method")
if not isinstance(method, str) or not method.strip():
return False, ErrorHandler.create_invalid_request_error(
request_data.get("id")
)
return True, None
@staticmethod
def validate_prefecture_param(params):
"""
県名パラメータの妥当性を検証する
Args:
params (dict): パラメータ
Returns:
tuple: (is_valid, error_message)
- is_valid (bool): 妥当性
- error_message (str or None): エラーメッセージ（無効な場合）
"""
if not isinstance(params, dict):
return False, "パラメータは辞書形式である必要があります"
if "prefecture" not in params:
return False, "prefectureパラメータが指定されていません"
prefecture = params.get("prefecture")
if not isinstance(prefecture, str):
return False, "prefectureパラメータは文字列である必要があります"
if not prefecture.strip():
return False, "prefectureパラメータが空です"
if len(prefecture) > 50:
return False, "prefectureパラメータが長すぎます（最大50文字）"
return True, None
実行手順
サーバー起動
python3のインストール手順は省きます。
まずは以下のコマンドで環境構築します。
pip install -r requirements.txt
その後、以下のコマンドでサーバーを起動します。
python3 main.py
確認
ローカルサーバーから以下のコマンドを実行することで、確認できます。
curl -X POST http://localhost:8000/ \
-H "Content-Type: application/json" \
-d '{
"jsonrpc": "2.0",
"method": "get_tourist_spot",
"params": {"prefecture": "東京都"},
"id": 1
}'
以下のようなレスポンスがあれば成功です！
"jsonrpc": "2.0",
"result": {
"prefecture": "東京都",
"tourist_spot": "浅草寺 - 雷門や仲見世通りで賑わう東京の名所です。"
},
"id": 1
コメント