data-collector-mcp/tools/test_mcp_call.py

from __future__ import annotations

import argparse
import asyncio
import json
import os
from typing import Any

from fastmcp import Client


def _json_default(value: Any) -> Any:
    if hasattr(value, "model_dump"):
        return value.model_dump()
    if hasattr(value, "dict"):
        return value.dict()
    if hasattr(value, "__dict__"):
        return value.__dict__
    return str(value)


def _parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Call a data-collector-mcp tool over HTTP.")
    parser.add_argument(
        "--url",
        default=os.getenv("MCP_TEST_URL", "http://127.0.0.1:8501/mcp"),
        help="MCP HTTP URL. Default: http://127.0.0.1:8501/mcp",
    )
    parser.add_argument(
        "--tool",
        default="project.list",
        help="Tool name to call. Default: project.list",
    )
    parser.add_argument(
        "--args",
        default="{}",
        help='Tool arguments as JSON. Example: --args "{\"project_key\":\"dev-01\"}"',
    )
    parser.add_argument(
        "--list-tools",
        action="store_true",
        help="List available tools instead of calling one tool.",
    )
    return parser.parse_args()


async def _main() -> None:
    args = _parse_args()
    try:
        tool_args = json.loads(args.args)
    except json.JSONDecodeError as exc:
        raise SystemExit(f"invalid --args JSON: {exc}") from exc
    if not isinstance(tool_args, dict):
        raise SystemExit("--args must be a JSON object")

    async with Client(args.url) as client:
        if args.list_tools:
            print(json.dumps({"request": {"url": args.url, "action": "list_tools"}}, ensure_ascii=False, indent=2))
            result = await client.list_tools()
        else:
            print(
                json.dumps(
                    {
                        "request": {
                            "url": args.url,
                            "tool": args.tool,
                            "arguments": tool_args,
                        }
                    },
                    ensure_ascii=False,
                    indent=2,
                    default=_json_default,
                )
            )
            result = await client.call_tool(args.tool, tool_args)

    print(json.dumps({"response": result}, ensure_ascii=False, indent=2, default=_json_default))


if __name__ == "__main__":
    asyncio.run(_main())