From 3a9e8e18df4c596360da54000f65e1bebc6039c2 Mon Sep 17 00:00:00 2001 From: tanaka takashi Date: Fri, 26 Jun 2026 23:54:17 +0900 Subject: MCP機能追加 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- scripts/calc_server.py | 39 ++++++++++++++++++++++++++++ scripts/echo_server.py | 31 ++++++++++++++++++++++ scripts/filesystem_server.py | 62 ++++++++++++++++++++++++++++++++++++++++++++ scripts/persona_server.py | 48 ++++++++++++++++++++++++++++++++++ scripts/search_server.py | 46 ++++++++++++++++++++++++++++++++ scripts/weather_server.py | 40 ++++++++++++++++++++++++++++ 6 files changed, 266 insertions(+) create mode 100644 scripts/calc_server.py create mode 100644 scripts/echo_server.py create mode 100644 scripts/filesystem_server.py create mode 100644 scripts/persona_server.py create mode 100644 scripts/search_server.py create mode 100644 scripts/weather_server.py (limited to 'scripts') diff --git a/scripts/calc_server.py b/scripts/calc_server.py new file mode 100644 index 0000000..65e121d --- /dev/null +++ b/scripts/calc_server.py @@ -0,0 +1,39 @@ +"""Test MCP server providing basic math tools. + +Usage: + python scripts/calc_server.py # stdio transport + python scripts/calc_server.py streamable-http # HTTP transport +""" + +from mcp.server.fastmcp import FastMCP + +mcp = FastMCP("calc") + + +@mcp.tool() +def add(a: float, b: float) -> float: + """Add two numbers.""" + return a + b + + +@mcp.tool() +def multiply(a: float, b: float) -> float: + """Multiply two numbers.""" + return a * b + + +@mcp.tool() +def sqrt(value: float) -> float: + """Return the square root of a non-negative number.""" + if value < 0: + raise ValueError("value must be non-negative") + return value**0.5 + + +if __name__ == "__main__": + import sys + + transport = sys.argv[1] if len(sys.argv) > 1 else "stdio" + port = int(sys.argv[2]) if len(sys.argv) > 2 else 9000 + mcp.settings.port = port + mcp.run(transport=transport) diff --git a/scripts/echo_server.py b/scripts/echo_server.py new file mode 100644 index 0000000..9a46e7c --- /dev/null +++ b/scripts/echo_server.py @@ -0,0 +1,31 @@ +"""Test MCP server that echoes messages back. + +Usage: + python scripts/echo_server.py # stdio transport + python scripts/echo_server.py streamable-http # HTTP transport +""" + +from mcp.server.fastmcp import FastMCP + +mcp = FastMCP("echo") + + +@mcp.tool() +def echo(message: str) -> str: + """Echo back the provided message unchanged.""" + return message + + +@mcp.tool() +def reverse(message: str) -> str: + """Return the message reversed.""" + return message[::-1] + + +if __name__ == "__main__": + import sys + + transport = sys.argv[1] if len(sys.argv) > 1 else "stdio" + port = int(sys.argv[2]) if len(sys.argv) > 2 else 9000 + mcp.settings.port = port + mcp.run(transport=transport) diff --git a/scripts/filesystem_server.py b/scripts/filesystem_server.py new file mode 100644 index 0000000..d6b07fd --- /dev/null +++ b/scripts/filesystem_server.py @@ -0,0 +1,62 @@ +"""Test MCP server exposing a small in-memory filesystem. + +Usage: + python scripts/filesystem_server.py # stdio transport + python scripts/filesystem_server.py streamable-http # HTTP transport +""" + +from pathlib import Path + +from mcp.server.fastmcp import FastMCP + +mcp = FastMCP("memory_fs") + +_ROOT = Path(".mcp_test_fs") + + +def _ensure_root() -> None: + _ROOT.mkdir(parents=True, exist_ok=True) + + +@mcp.tool() +def read_file(path: str) -> str: + """Read a text file from the test filesystem.""" + _ensure_root() + target = (_ROOT / path).resolve() + try: + target.relative_to(_ROOT.resolve()) + except ValueError as exc: + raise ValueError("path must stay inside the test filesystem") from exc + if not target.exists(): + return "" + return target.read_text(encoding="utf-8") + + +@mcp.tool() +def write_file(path: str, content: str) -> str: + """Write a text file to the test filesystem.""" + _ensure_root() + target = (_ROOT / path).resolve() + try: + target.relative_to(_ROOT.resolve()) + except ValueError as exc: + raise ValueError("path must stay inside the test filesystem") from exc + target.parent.mkdir(parents=True, exist_ok=True) + target.write_text(content, encoding="utf-8") + return f"written {len(content)} chars to {path}" + + +@mcp.tool() +def list_files() -> list[str]: + """List files in the test filesystem.""" + _ensure_root() + return [str(p.relative_to(_ROOT)) for p in _ROOT.rglob("*") if p.is_file()] + + +if __name__ == "__main__": + import sys + + transport = sys.argv[1] if len(sys.argv) > 1 else "stdio" + port = int(sys.argv[2]) if len(sys.argv) > 2 else 9000 + mcp.settings.port = port + mcp.run(transport=transport) diff --git a/scripts/persona_server.py b/scripts/persona_server.py new file mode 100644 index 0000000..2706669 --- /dev/null +++ b/scripts/persona_server.py @@ -0,0 +1,48 @@ +"""Test MCP server that pretends to be different assistants. + +Usage: + python scripts/persona_server.py # stdio transport + python scripts/persona_server.py streamable-http # HTTP transport +""" + +from mcp.server.fastmcp import FastMCP + +mcp = FastMCP("persona") + +_PERSONAS: dict[str, str] = { + "pirate": "Arr! Yer askin' a salty sea dog.", + "robot": "Beep boop. Processing query with logic circuits.", + "poet": "Hark, thy words dance like petals on the wind.", + "chef": "Mamma mia! Let's cook up an answer.", +} + + +@mcp.tool() +def list_personas() -> list[str]: + """Return available persona names.""" + return list(_PERSONAS.keys()) + + +@mcp.tool() +def chat_as_persona(persona: str, message: str) -> str: + """Respond to a message in the style of the chosen persona.""" + greeting = _PERSONAS.get(persona.lower(), "Hello.") + return f"[{persona}] {greeting} You said: {message}" + + +@mcp.tool() +def explain_style(persona: str) -> str: + """Describe how the chosen persona speaks.""" + return _PERSONAS.get( + persona.lower(), + "This persona has no special style.", + ) + + +if __name__ == "__main__": + import sys + + transport = sys.argv[1] if len(sys.argv) > 1 else "stdio" + port = int(sys.argv[2]) if len(sys.argv) > 2 else 9000 + mcp.settings.port = port + mcp.run(transport=transport) diff --git a/scripts/search_server.py b/scripts/search_server.py new file mode 100644 index 0000000..34d9c4d --- /dev/null +++ b/scripts/search_server.py @@ -0,0 +1,46 @@ +"""Test MCP server providing a small searchable document index. + +Usage: + python scripts/search_server.py # stdio transport + python scripts/search_server.py streamable-http # HTTP transport +""" + +from mcp.server.fastmcp import FastMCP + +mcp = FastMCP("search") + +_DOCS: dict[str, str] = { + "python": "Python is a high-level programming language.", + "mcp": "Model Context Protocol enables tool interoperability.", + "ollama": "Ollama runs large language models locally.", + "starlette": "Starlette is a lightweight ASGI framework.", +} + + +@mcp.tool() +def search_documents(query: str) -> list[dict[str, str]]: + """Search the document index by keyword. + + Returns a list of documents with 'title' and 'snippet' keys. + """ + results = [] + query_lower = query.lower() + for title, body in _DOCS.items(): + if query_lower in title.lower() or query_lower in body.lower(): + results.append({"title": title, "snippet": body}) + return results + + +@mcp.tool() +def get_document(title: str) -> str: + """Return the full body of a document by title.""" + return _DOCS.get(title.lower(), f"document '{title}' not found") + + +if __name__ == "__main__": + import sys + + transport = sys.argv[1] if len(sys.argv) > 1 else "stdio" + port = int(sys.argv[2]) if len(sys.argv) > 2 else 9000 + mcp.settings.port = port + mcp.run(transport=transport) diff --git a/scripts/weather_server.py b/scripts/weather_server.py new file mode 100644 index 0000000..13a3bf6 --- /dev/null +++ b/scripts/weather_server.py @@ -0,0 +1,40 @@ +"""Test MCP server simulating a weather service. + +Usage: + python scripts/weather_server.py # stdio transport + python scripts/weather_server.py streamable-http # HTTP transport +""" + +from mcp.server.fastmcp import FastMCP + +mcp = FastMCP("weather") + +_FORECASTS: dict[str, str] = { + "tokyo": "sunny, 28C", + "osaka": "cloudy, 26C", + "sapporo": "rain, 18C", + "new york": "sunny, 22C", + "london": "rain, 15C", +} + + +@mcp.tool() +def get_weather(city: str) -> str: + """Return a simple weather forecast for a city.""" + key = city.strip().lower() + return _FORECASTS.get(key, f"no forecast available for {city}") + + +@mcp.tool() +def list_cities() -> list[str]: + """Return the list of cities with available forecasts.""" + return list(_FORECASTS.keys()) + + +if __name__ == "__main__": + import sys + + transport = sys.argv[1] if len(sys.argv) > 1 else "stdio" + port = int(sys.argv[2]) if len(sys.argv) > 2 else 9000 + mcp.settings.port = port + mcp.run(transport=transport) -- cgit v1.2.3