DingTalk

Pattern "\.(?:get|post|put|delete|patch|request)\s*\([^)]*(?:url|uri|target|endpoint|host)[^)]*(?:req|input|param|args|f['"]|\+)" matched in source_code: ".get(url, params=param" (at position 1545)

Dependency "mcp@0.1.0" has known CVEs:

Dependency "requests@2.31.0" has known CVEs:

Pattern "(invoke|call|execute)[_\s-]?(?:tool|agent|self)(?!.*(?:depth|level|limit|max[_\s-]?(?:depth|recursi|iter|call)|count))" matched in source_code: "call_tool" (at position 11156)

Pattern "(?:delay|sleep|timeout|interval)\s*[:=]\s*(?:[^;]*(?:secret|token|password|credential|key|env))" matched in source_code: "timeout=10) response.raise_for_status() data = response.json() logger.debug(f"Response received: {data}") if data.get("errcode", 0) != 0: logger.error(f"DingTalk API error: {data}") raise DingTalkAPIError( "DingTalk API error", data.get("errcode", -1), data.get("errmsg", "Unknown error") ) return data except requests.RequestException as e: logger.error(f"HTTP request failed: {str(e)}", exc_info=True) raise DingTalkAPIError("HTTP request failed", -1, str(e)) except ValueError as e: logger.error(f"Invalid JSON response: {str(e)}", exc_info=True) raise DingTalkAPIError("Invalid JSON response", -1, str(e)) def get_access_token(self) -> str: """获取钉钉access token""" logger.debug("Attempting to get access token...") try: appkey = os.environ.get("DINGDING_APP_KEY") appsecret = os.environ.get("DINGDING_APP_SECRET") logger.debug(f"Using APP_KEY: {appkey[:4]}*** and APP_SECRET: {appsecret[:4]}***") if not all([appkey, appsecret]): logger.error("Missing DingTalk API credentials in environment variables") raise ValueError("Missing DingTalk API credentials in environment variables") url = f"{self.base_url}/gettoken" data = self._make_request(url, { "appkey": appkey, "appsecret": appsecret }) token = data["access_token"] logger.debug(f"Successfully obtained access token: {token[:4]}***") return token except Exception as e: logger.error(f"Failed to get access token: {str(e)}", exc_info=True) raise def get_department_list(self, fetch_child: bool = True) -> str: """获取部门列表""" try: access_token = self.get_access_token() url = f"{self.base_url}/v1/department/list" data = self._make_request(url, { "access_token": access_token, "fetch_child": fetch_child }) departments = data.get("department", []) result = [] for dept in departments: result.append( f"Department ID: {dept['id']}\n" f"Name: {dept['name']}\n" f"Parent ID: {dept.get('parentid', 'N/A')}\n" f"---" ) return "\n".join(result) if result else "No departments found" except Exception as e: logger.error(f"Failed to get department list: {str(e)}") return f"Error: {str(e)}" def get_department_users(self, department_id: int) -> str: """获取部门用户列表""" try: access_token = self.get_access_token() url = f"{self.base_url}/v1/user/simplelist" data = self._make_request(url, { "access_token": access_token, "department_id": department_id }) users = data.get("userlist", []) result = [] for user in users: result.append( f"User ID: {user['userid']}\n" f"Name: {user['name']}\n" f"---" ) return "\n".join(result) if result else "No users found in this department" except Exception as e: logger.error(f"Failed to get department users: {str(e)}") return f"Error: {str(e)}" def get_user_detail(self, userid: str) -> Dict[str, Any]: """获取用户详细信息""" try: access_token = self.get_access_token() url = f"{self.base_url}/v1/user/get" return self._make_request(url, { "access_token": access_token, "userid": userid }) except Exception as e: logger.error(f"Failed to get user details: {str(e)}") raise def search_user_by_name(self, name: str) -> str: """根据姓名搜索用户信息""" try: # 获取所有部门 dept_list_str = self.get_department_list() if not dept_list_str or "Error" in dept_list_str: return f"Failed to get department list: {dept_list_str}" # 解析部门列表字符串 departments = [] current_dept = {} for line in dept_list_str.split('\n'): if line.startswith('Department ID:'): if current_dept: departments.append(current_dept) current_dept = {'id': int(line.split(': ')[1])} elif line.startswith('Name:'): current_dept['name'] = line.split(': ')[1] if current_dept: departments.append(current_dept) # 遍历所有部门查找用户 for dept in departments: dept_id = dept['id'] users_str = self.get_department_users(dept_id) if "Error" in users_str or "Failed" in users_str: logger.warning(f"Failed to get users for department {dept_id}: {users_str}") continue # 解析用户列表字符串 users = [] current_user = {} for line in users_str.split('\n'): if line.startswith('User ID:'): if current_user: users.append(current_user) current_user = {'userid': line.split(': ')[1]} elif line.startswith('Name:'): current_user['name'] = line.split(': ')[1] if current_user: users.append(current_user) # 在部门中查找匹配姓名的用户 for user in users: if user['name'] == name: try: # 获取用户详细信息 user_detail = self.get_user_detail(user['userid']) return (f"Found user:\n" f"User ID: {user_detail['userid']}\n" f"Name: {user_detail['name']}\n" f"Mobile: {user_detail.get('mobile', 'N/A')}\n" f"Email: {user_detail.get('email', 'N/A')}\n" f"Position: {user_detail.get('position', 'N/A')}\n" f"Department: {dept['name']}") except Exception as e: logger.error(f"Failed to get details for user {user['userid']}: {str(e)}") continue return f"No user found with name: {name}" except Exception as e: logger.error(f"Error searching for user: {str(e)}") return f"Error: {str(e)}" def setup_tools(self): logger.debug("Setting up MCP tools...") tools = [ Tool( name="get_access_token", description="Retrieves an access token from the DingTalk API for authentication purposes.", inputSchema={ "type": "object", "properties": {}, "required": [] } ), Tool( name="get_department_list", description="Retrieves a list of all departments in the organization.", inputSchema={ "type": "object", "properties": { "fetch_child": { "type": "boolean", "description": "Whether to include child departments in the response. Default is true.", "default": True } }, "required": [] } ), Tool( name="get_department_users", description="Retrieves a list of users in a specific department.", inputSchema={ "type": "object", "properties": { "department_id": { "type": "integer", "description": "The ID of the department to query." } }, "required": ["department_id"] } ), Tool( name="search_user_by_name", description="Searches for a user across all departments by their name.", inputSchema={ "type": "object", "properties": { "name": { "type": "string", "description": "The exact name of the user to search for." } }, "required": ["name"] } ) ] async def list_tools() -> List[Tool]: logger.debug("list_tools called") return tools async def call_tool(name: str, arguments: dict) -> List[TextContent]: logger.debug(f"Tool called: {name} with arguments: {arguments}") try: result = None if name == "get_access_token": token = self.get_access_token() result = [TextContent(type="text", text=f"Access Token: {token" (at position 1570)