fix(agent-scan): Tool.inputSchema renamed to input_schema in mcp 2.0

agent-scan/agent_scan/utils/mcp_tools.py

@@ -165,11 +165,11 @@ async def describe_mcp_tools(self) -> str:
         xml_lines = ["<mcp_tools>"]
         for t in data.tools:
             # 缓存工具 schema，用于后续参数类型转换
-            self._tools_schema[t.name] = t.inputSchema
+            self._tools_schema[t.name] = t.input_schema
 
             parameters = ''
-            for k, param in t.inputSchema['properties'].items():
-                required = 'true' if k in t.inputSchema.get("required", []) else 'false'
+            for k, param in t.input_schema['properties'].items():
+                required = 'true' if k in t.input_schema.get("required", []) else 'false'
                 param_type = param.get('type', 'string')
                 # 构建基础属性
                 base_attrs = f'name="{k}" type="{param_type}" required="{required}"'

agent-scan/pytests/test_mcp_sdk_v2.py

@@ -76,5 +76,53 @@ async def fake_streamable_http_client(url, *, http_client=None, terminate_on_clo
         self.assertTrue(_Session.instances[0].initialized)
 
 
+class _Tool:
+    """Mimics mcp 2.0 Tool: exposes input_schema (snake_case), no inputSchema."""
+
+    def __init__(self, name, description, input_schema):
+        self.name = name
+        self.description = description
+        self.input_schema = input_schema
+
+
+class _ListToolsResult:
+    def __init__(self, tools):
+        self.tools = tools
+
+
+class _FakeSession:
+    async def list_tools(self):
+        return _ListToolsResult([
+            _Tool(
+                "echo",
+                "Echo a message",
+                {
+                    "type": "object",
+                    "properties": {"message": {"type": "string"}},
+                    "required": ["message"],
+                },
+            )
+        ])
+
+
+class MCP20ToolSchemaTests(unittest.IsolatedAsyncioTestCase):
+    async def test_describe_mcp_tools_reads_input_schema(self):
+        manager = MCPTools("https://example.test/mcp", "streamable-http")
+
+        @asynccontextmanager
+        async def fake_session(_self):
+            yield _FakeSession()
+
+        with patch.object(MCPTools, "_session", fake_session):
+            description = await manager.describe_mcp_tools()
+
+        self.assertIn("echo", description)
+        self.assertIn('name="message" type="string" required="true"', description)
+        self.assertEqual(
+            manager._tools_schema["echo"]["properties"]["message"]["type"],
+            "string",
+        )
+
+
 if __name__ == "__main__":
     unittest.main()