107 lines
3.4 KiB
Python
107 lines
3.4 KiB
Python
#!/usr/bin/env python3
|
|
"""Small CLI for one-off SxCP MCP bridge calls.
|
|
|
|
The repository smoke tests run with the system Python, so MCP dependencies are
|
|
imported only after a network subcommand is selected. For live bridge calls, run
|
|
this with the Python environment that has the `mcp` package installed.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
from typing import Any
|
|
|
|
|
|
DEFAULT_BRIDGE_URL = "http://192.168.1.12:9188/mcp"
|
|
|
|
|
|
def _json_loads(value: str) -> dict[str, Any]:
|
|
try:
|
|
parsed = json.loads(value)
|
|
except json.JSONDecodeError as exc:
|
|
raise argparse.ArgumentTypeError(str(exc)) from exc
|
|
if not isinstance(parsed, dict):
|
|
raise argparse.ArgumentTypeError("arguments JSON must decode to an object")
|
|
return parsed
|
|
|
|
|
|
def _json_default(value: Any) -> Any:
|
|
if hasattr(value, "model_dump"):
|
|
return value.model_dump(mode="json")
|
|
if hasattr(value, "__dict__"):
|
|
return value.__dict__
|
|
return str(value)
|
|
|
|
|
|
async def _list_tools(bridge_url: str) -> int:
|
|
from mcp.client.session import ClientSession
|
|
from mcp.client.streamable_http import streamablehttp_client
|
|
|
|
async with streamablehttp_client(bridge_url) as (read, write, _):
|
|
async with ClientSession(read, write) as session:
|
|
await session.initialize()
|
|
tools = (await session.list_tools()).tools
|
|
for tool in tools:
|
|
print(tool.name)
|
|
return 0
|
|
|
|
|
|
async def _call_tool(bridge_url: str, tool_name: str, arguments: dict[str, Any]) -> int:
|
|
from mcp.client.session import ClientSession
|
|
from mcp.client.streamable_http import streamablehttp_client
|
|
|
|
async with streamablehttp_client(bridge_url) as (read, write, _):
|
|
async with ClientSession(read, write) as session:
|
|
await session.initialize()
|
|
result = await session.call_tool(tool_name, arguments)
|
|
print(json.dumps(result, ensure_ascii=True, indent=2, default=_json_default))
|
|
return 0
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument(
|
|
"--bridge-url",
|
|
default=DEFAULT_BRIDGE_URL,
|
|
help=f"MCP bridge URL. Default: {DEFAULT_BRIDGE_URL}",
|
|
)
|
|
subparsers = parser.add_subparsers(dest="command", required=True)
|
|
|
|
subparsers.add_parser("list-tools", help="List available MCP tool names.")
|
|
|
|
call_parser = subparsers.add_parser("call-tool", help="Call one MCP tool.")
|
|
call_parser.add_argument("tool_name", help="Tool name to call.")
|
|
call_parser.add_argument(
|
|
"--arguments-json",
|
|
type=_json_loads,
|
|
default={},
|
|
help="JSON object with tool arguments.",
|
|
)
|
|
return parser
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = build_parser()
|
|
args = parser.parse_args(argv)
|
|
|
|
try:
|
|
import anyio
|
|
except ImportError as exc:
|
|
raise SystemExit(
|
|
"sxcp_mcp_client requires the MCP Python environment for network calls; "
|
|
"try /media/p5/miniforge3/bin/python."
|
|
) from exc
|
|
|
|
if args.command == "list-tools":
|
|
return anyio.run(_list_tools, args.bridge_url)
|
|
if args.command == "call-tool":
|
|
return anyio.run(_call_tool, args.bridge_url, args.tool_name, args.arguments_json)
|
|
parser.error(f"unknown command: {args.command}")
|
|
return 2
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|