Skip to content

Instantly share code, notes, and snippets.

@kohya-ss

kohya-ss/client_v4.py

Last active July 2, 2025 12:56
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save kohya-ss/ad428205329dfeacd92141a88715530f to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save kohya-ss/ad428205329dfeacd92141a88715530f to your computer and use it in GitHub Desktop.
Download ZIP
メッセージボックスMCPサーバ
Raw
client_v4.py
# client_v4.py
"""
Message Box MCP Client
インストール方法等はserver_v4.pyを参照。
## 使用法
メッセージ取得:
python client_v4.py --tool get_message --address test
メッセージ送信:
python client_v4.py --tool post_message --address test --message "Hello, MCP!"
"""
import asyncio
from fastmcp import Client, FastMCP
# HTTP server
config = {
"mcpServers": {
"localhost": {
"url": "http://localhost:8000/mcp",
"transport": "streamable-http",
},
}
}
client = Client(config)
async def main(tool: str = None, address: str = "test", message: str = "Hello, MCP!"):
"""Main function to run the MCP client."""
async with client:
# Basic server interaction
await client.ping()
# # List available operations
# tools = await client.list_tools()
# resources = await client.list_resources()
# prompts = await client.list_prompts()
# # Execute operations
# print("Available tools:", tools)
# print("Available resources:", resources)
# print("Available prompts:", prompts)
print(f"Calling tool: {tool or 'get_message'} with address: {address} and message: {message}")
args = {"address": address}
if tool == "post_message":
args["message"] = message
result = await client.call_tool(name=tool or "get_message", arguments=args)
print(result)
if __name__ == "__main__":
import argparse
parser = argparse.ArgumentParser(description="Run the Message Box MCP client.")
parser.add_argument("--tool", type=str, default=None, help="Tool to call (e.g., 'get_message')")
parser.add_argument("--address", type=str, default="test", help="Address to send messages to")
parser.add_argument("--message", type=str, default="Hello, MCP!", help="Message to send")
args = parser.parse_args()
asyncio.run(main(args.tool, args.address, args.message))
Raw
server_v4.py
# server_v4.py FastMCP 2.0 version
"""
Message Box Server
ポート番号は末尾の行に記載されています。
現在のメッセージボックスの中身と、ログがカレントディレクトリに保存されます。
## インストール
適切な仮想環境で、 `pip install fastmcp`
## 起動
`python server_v4.py`
## Gemini CLIの設定
{
"mcpServers": {
"messagebox": {
"httpUrl": "http://localhost:8000/mcp",
"timeout": 5000
}
}
}
WSLからWindowsのFastMCPサーバーにアクセスする場合は、Windows側のIPアドレスを指定する必要があります。（例: 192.168.0.xxx など）
client_v4.pyを実行して、コマンドラインから任意のaddressにメッセージを送信できます。
"""
from contextlib import asynccontextmanager
import contextlib
import json
import os
from pathlib import Path
from fastmcp import FastMCP
from typing import Optional
mcp = FastMCP(
name="Message Box MCP Server",
instructions="""
このサーバーは、シンプルなメッセージボックスを提供します。
メッセージを保存し、取得し、削除することができます。
""",
)
# ファイルパスを設定
MESSAGE_FILE = Path(__file__).parent / "message_box.json"
LOG_FILE = Path(__file__).parent / "message_box.log"
message_box: dict[str, str] = {}
def load_messages():
"""ファイルからメッセージを読み込む"""
global message_box
if MESSAGE_FILE.exists():
try:
with open(MESSAGE_FILE, "r", encoding="utf-8") as f:
message_box = json.load(f)
print(f"Loaded {len(message_box)} messages from file.")
except (json.JSONDecodeError, IOError) as e:
print(f"Error loading messages: {e}")
message_box = {}
else:
message_box = {}
print("No existing message file found. Starting with empty message box.")
def save_messages():
"""メッセージをファイルに保存する"""
try:
with open(MESSAGE_FILE, "w", encoding="utf-8") as f:
json.dump(message_box, f, ensure_ascii=False, indent=2)
print(f"Saved {len(message_box)} messages to file.")
except IOError as e:
print(f"Error saving messages: {e}")
@mcp.tool(description="メッセージボックスにメッセージを保存します。")
def post_message(address: str, message: str) -> str:
"""
新しいメッセージを受け取り、メッセージボックスを上書きします。
"""
global message_box
message_box[address] = message
save_messages()
# ログファイルにメッセージを保存
if not LOG_FILE.exists():
with open(LOG_FILE, "w", encoding="utf-8") as log_file:
log_file.write("Message Log:\n")
with open(LOG_FILE, "a", encoding="utf-8") as log_file:
log_file.write(f"{address}: {message}\n")
print(f"Received message for {address}: {message}")
return "メッセージを預かりました。"
@mcp.tool(description="現在のメッセージボックスの内容を取得します。メッセージがなければ空文字列を返します。")
def get_message(address: str) -> str:
"""
現在のメッセージボックスの内容を返します。
メッセージがなければ空文字列を返します。
"""
if address not in message_box:
print("No message in the box.")
return ""
print(f"Current message for {address}: {message_box[address]}")
return message_box[address]
@mcp.tool(description="現在のメッセージボックスを空にします。")
def delete_message(address: str) -> str:
"""
現在のメッセージボックスを空にします。
一連の処理が終わった後に呼び出すと安全です。
"""
global message_box
message_box.pop(address, None)
save_messages()
print(f"Message box for {address} cleared.")
return "メッセージを削除しました。"
load_messages() # サーバー起動時にメッセージを読み込む
if __name__ == "__main__":
mcp.run(transport="http", host="0.0.0.0", port=8000)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment