weather_client.py•5.05 kB
"""
weather_client.py
MCP stdio client for the weather server in weather_server.py.
Usage examples:
- Default (spawns server and runs a quick demo):
python weather_client.py --location "San Francisco" --days 2
- Interactive ping and tool listing only:
python weather_client.py --list-only
- Custom server command/args (if needed):
python weather_client.py --cmd python --cmd-arg weather_server.py
Notes:
- Requires the `mcp` library (added to requirements.txt).
- This client launches the MCP server via stdio and calls tools.
"""
from __future__ import annotations
import argparse
import asyncio
import json
import sys
from typing import Any
def _jsonify(obj: Any) -> Any:
"""Best-effort conversion to something JSON-serializable."""
if obj is None or isinstance(obj, (str, int, float, bool)):
return obj
if isinstance(obj, (list, tuple)):
return [_jsonify(x) for x in obj]
if isinstance(obj, dict):
return {str(k): _jsonify(v) for k, v in obj.items()}
# Try common object patterns
for attr in ("model_dump", "dict", "_asdict"):
if hasattr(obj, attr) and callable(getattr(obj, attr)):
try:
return _jsonify(getattr(obj, attr)())
except Exception:
pass
if hasattr(obj, "__dict__"):
try:
return _jsonify(vars(obj))
except Exception:
pass
return str(obj)
async def run_client(args: argparse.Namespace) -> int:
try:
# Lazy imports so error messages are clearer if missing
from mcp.client.session import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client
except Exception as e: # pragma: no cover - import-time failure messaging
print(
"Error: the 'mcp' package is required to run the client.\n"
"Install it with: pip install mcp",
file=sys.stderr,
)
print(f"Import failure detail: {e}", file=sys.stderr)
return 2
# Default to spawning the local weather_server.py
server_cmd = args.cmd or sys.executable
server_args = args.cmd_arg or ["weather_server.py"]
params = StdioServerParameters(command=server_cmd, args=server_args)
# Connect to the MCP server via stdio
async with stdio_client(params) as (read, write):
async with ClientSession(read, write) as session:
# Initialize the session and get server capabilities
await session.initialize()
# List tools
tools_resp = await session.list_tools()
tools_list = getattr(tools_resp, "tools", [])
print("Available tools:")
for t in tools_list:
name = getattr(t, "name", None) or getattr(t, "id", None) or "<unknown>"
desc = getattr(t, "description", "")
print(f"- {name}: {desc}")
# Try ping tool (optional)
try:
ping_result = await session.call_tool("ping", arguments={})
print("\nPing result:")
print(json.dumps(_jsonify(ping_result), indent=2))
except Exception:
# If no ping tool, ignore quietly
pass
if args.list_only:
return 0
# If a location is provided, call get_forecast
if args.location:
call_args: dict[str, Any] = {"location": args.location}
if args.days is not None:
call_args["days"] = args.days
print("\nCalling get_forecast with:", call_args)
try:
forecast = await session.call_tool("get_forecast", arguments=call_args)
except Exception as e:
print(f"Error calling get_forecast: {e}", file=sys.stderr)
return 1
print("\nForecast result:")
print(json.dumps(_jsonify(forecast), indent=2))
else:
print("\nTip: pass --location 'City' [--days N] to fetch a forecast.")
return 0
def build_parser() -> argparse.ArgumentParser:
p = argparse.ArgumentParser(description="MCP client for weather_server.py")
p.add_argument(
"--cmd",
help="Server command to run (default: current Python)",
default=None,
)
p.add_argument(
"--cmd-arg",
action="append",
help="Argument to pass to the server command. Can be repeated.",
)
p.add_argument("--location", help="Location/city to query", default=None)
p.add_argument("--days", type=int, help="Days for forecast (1-7)")
p.add_argument(
"--list-only",
action="store_true",
help="Only list tools and attempt ping; do not call forecast.",
)
return p
def main(argv: list[str] | None = None) -> int:
parser = build_parser()
args = parser.parse_args(argv)
return asyncio.run(run_client(args))
if __name__ == "__main__":
raise SystemExit(main())