SmartBizTrip/tests/test_weather_tool_parameters.py

"""验证天气子助手能将多个命名参数传给 MCP 工具，不访问外部服务。"""

import unittest
from unittest.mock import Mock, patch

from langchain_core.tools import Tool

from a2a_server import weather_server


class WeatherToolTests(unittest.IsolatedAsyncioTestCase):
    async def test_city_and_date_range_reach_mcp_as_named_parameters(self):
        parameters = {
            "city": "北京",
            "start_date": "2026-09-26",
            "end_date": "2026-09-28",
        }
        mcp_call = Mock(return_value="北京天气查询结果")
        # 复现 python-a2a 0.5.4 适配器返回的单输入 Tool。
        adapter_tool = Tool(
            name="query_weather",
            description="查询指定城市和日期范围的天气",
            func=mcp_call,
            metadata={"parameters": {
                "type": "object",
                "properties": {key: {"type": "string"} for key in parameters},
                "required": list(parameters),
            }},
        )

        class ToolCallingExecutor:
            def __init__(self, *, agent, tools, verbose):
                self.tool = tools[0]

            async def ainvoke(self, inputs):
                # 模拟模型的结构化工具调用，使用真实 LangChain 参数分发逻辑。
                return {"output": await self.tool.ainvoke(parameters)}

        with (
            patch.object(weather_server, "to_langchain_tool", return_value=[adapter_tool]),
            patch.object(weather_server, "create_tool_calling_agent"),
            patch.object(weather_server, "AgentExecutor", ToolCallingExecutor),
        ):
            result = await weather_server.query_weather("北京未来三天天气")

        self.assertEqual(result, {"status": "success", "message": "北京天气查询结果"})
        mcp_call.assert_called_once_with(**parameters)


if __name__ == "__main__":
    unittest.main()