Contract Inspector MCP

Instructions

Implementation Reference

• contract_inspector/main.py:24-67 (handler)

  The contract_info tool handler function, registered via @mcp.tool(). It validates the input contract address, calls ContractAnalyzer.analyze_contract(), formats the result as JSON, and handles errors.
  @mcp.tool() async def contract_info(contract_address: str): """ 获取 EVM 合约的完整链上信息，包括调用所有无参数的 view 函数 Args: contract_address: EVM 合约地址 (0x开头的42位十六进制字符串) Returns: 合约的详细信息，包括基础信息和所有view函数的调用结果 """ # 验证合约地址格式 if not contract_address or not isinstance(contract_address, str): raise ValueError("合约地址不能为空且必须是字符串") if not contract_address.startswith("0x") or len(contract_address) != 42: raise ValueError("合约地址格式无效，必须是0x开头的42位十六进制字符串") try: print(f"🔍 开始分析合约: {contract_address}") # 执行合约分析 result = await analyzer.analyze_contract(contract_address) # 格式化输出 formatted_result = json.dumps(result, ensure_ascii=False, indent=2) print(f"✅ 合约分析完成: {contract_address}") return formatted_result except Exception as e: error_msg = f"合约分析失败: {str(e)}" print(f"❌ {error_msg}") error_response = { "status": "error", "error": error_msg, "contract_address": contract_address, "tool": "contract_info" } return json.dumps(error_response, ensure_ascii=False, indent=2)
• contract_inspector/contract_analyzer.py:47-179 (helper)

  The core helper method implementing the contract analysis logic: validates address, fetches ABI, extracts and calls view functions, formats results.
  async def analyze_contract(self, contract_address: str) -> Dict[str, Any]: """ 分析合约信息 - 主要业务逻辑 Args: contract_address: 合约地址 Returns: Dict: 分析结果 """ # 初始化检查 if not await self.initialize(): return { "status": "error", "error": "分析器初始化失败", "timestamp": datetime.now().isoformat() } # 步骤1: 验证地址格式 if not is_valid_ethereum_address(contract_address): return { "status": "error", "error": f"无效的以太坊地址格式: {contract_address}", "timestamp": datetime.now().isoformat() } # 转换为校验和格式 checksum_address = to_checksum_address(contract_address) # 步骤2: 检查是否为合约地址 is_contract = await self.web3_client.is_contract_address(checksum_address) if not is_contract: return { "status": "error", "error": f"地址 {checksum_address} 不是合约地址", "timestamp": datetime.now().isoformat() } # 步骤3: 从 Etherscan 获取合约 ABI print(f"📡 正在获取合约 {checksum_address} 的 ABI...") abi = await self.etherscan_client.get_contract_abi(checksum_address) if not abi: return { "status": "error", "error": f"无法获取合约 ABI，可能合约未验证", "contract_address": checksum_address, "timestamp": datetime.now().isoformat() } # 步骤4: 筛选无参数的 view 函数 view_functions = extract_view_functions(abi) if not view_functions: return { "status": "warning", "message": "合约中没有找到无参数的 view 函数", "contract_address": checksum_address, "total_functions": len([f for f in abi if f.get("type") == "function"]), "timestamp": datetime.now().isoformat() } print(f"🔍 找到 {len(view_functions)} 个无参数的 view 函数") # 步骤5: 创建合约实例 contract = await self.web3_client.get_contract_instance(checksum_address, abi) if not contract: return { "status": "error", "error": "创建合约实例失败", "contract_address": checksum_address, "timestamp": datetime.now().isoformat() } # 步骤6: 批量调用 view 函数 function_names = [func["name"] for func in view_functions] print(f"🚀 正在调用 {len(function_names)} 个函数...") call_results = await self.web3_client.batch_call_view_functions( contract, function_names ) # 步骤7: 处理和格式化结果 successful_calls = [] failed_calls = [] for i, result in enumerate(call_results): if result["status"] == "success": # 获取函数的输出类型信息 func_info = view_functions[i] output_type = self._get_output_type(func_info) formatted_result = { "function_name": result["function_name"], "result": result["result"], "type": output_type, "status": "success" } # 格式化特殊类型的值 if output_type.startswith("uint") and isinstance(result["result"], int): if result["result"] > 10**15: # 可能是 wei 值 formatted_result["formatted_value"] = self._format_large_number(result["result"]) elif output_type == "address": formatted_result["checksum_address"] = to_checksum_address(str(result["result"])) successful_calls.append(formatted_result) else: failed_calls.append({ "function_name": result["function_name"], "error": result.get("error", "未知错误"), "status": "failed" }) # 获取额外的合约信息 contract_name = await self.etherscan_client.get_contract_name(checksum_address) # 组装最终结果 final_result = { "status": "success", "contract_address": checksum_address, "contract_name": contract_name, "analysis_summary": { "total_view_functions": len(view_functions), "successful_calls": len(successful_calls), "failed_calls": len(failed_calls) }, "successful_functions": successful_calls, "failed_functions": failed_calls if failed_calls else None, "timestamp": datetime.now().isoformat() } print(f"✅ 分析完成: 成功调用 {len(successful_calls)} 个函数，失败 {len(failed_calls)} 个") return final_result