Coverage for session_buddy / tools / team_tools.py: 19.75%

1#!/usr/bin/env python3 

2"""Team collaboration tools for session-mgmt-mcp. 

3 

4Following crackerjack architecture patterns for knowledge sharing, 

5team coordination, and collaborative development workflows. 

6 

7Refactored to use utility modules for reduced code duplication. 

8""" 

9 

10from __future__ import annotations 

11 

12import typing as t 

13from typing import TYPE_CHECKING, Any 

14 

15from session_buddy.utils.error_handlers import _get_logger 

16from session_buddy.utils.messages import ToolMessages 

17 

18if TYPE_CHECKING: 

19 from fastmcp import FastMCP 

20 

21 

22# Constants for error messages 

23TEAM_NOT_AVAILABLE_MSG = ( 

24 "Team collaboration features not available. Install optional dependencies" 

25) 

26 

27 

28# ============================================================================ 

29# Service Resolution 

30# ============================================================================ 

31 

32 

33async def _require_team_manager() -> Any: 

34 """Get team knowledge manager or raise with helpful error message.""" 

35 try: 

36 from session_buddy.team_knowledge import TeamKnowledgeManager 

37 

38 return TeamKnowledgeManager() 

39 except ImportError: 

40 _get_logger().warning("Team knowledge system not available") 

41 raise RuntimeError(TEAM_NOT_AVAILABLE_MSG) 

42 

43 

44async def _execute_team_operation( 

45 operation_name: str, 

46 operation: t.Callable[[Any], t.Awaitable[str]], 

47) -> str: 

48 """Execute a team operation with error handling.""" 

49 try: 

50 manager = await _require_team_manager() 

51 return await operation(manager) 

52 except RuntimeError as e: 

53 return f"❌ {e!s}" 

54 except ValueError as e: 

55 return f"❌ {operation_name} failed: {e!s}" 

56 except Exception as e: 

57 _get_logger().exception(f"Error in {operation_name}: {e}") 

58 return ToolMessages.operation_failed(operation_name, e) 

59 

60 

61# ============================================================================ 

62# Output Formatting Helpers 

63# ============================================================================ 

64 

65 

66def _format_search_result(result: dict[str, Any], index: int) -> str: 

67 """Format a single search result.""" 

68 output = f"**{index}.** " 

69 

70 # Add metadata 

71 if result.get("team_id"): 

72 output += f"[{result['team_id']}] " 

73 if result.get("author"): 

74 output += f"by {result['author']} " 

75 if result.get("timestamp"): 

76 output += f"({result['timestamp']}) " 

77 

78 # Add content preview 

79 content = result.get("content", "") 

80 output += f"\n{content[:200]}...\n" 

81 

82 # Add tags if available 

83 if result.get("tags"): 

84 output += f"🏷️ Tags: {', '.join(result['tags'])}\n" 

85 

86 # Add voting info if available 

87 if result.get("votes"): 

88 votes = result["votes"] 

89 output += f"👍 Votes: {votes} " 

90 

91 output += "\n" 

92 return output 

93 

94 

95def _format_search_scope(team_id: str | None, project_id: str | None) -> str: 

96 """Format the search scope string.""" 

97 search_scope = "team knowledge" 

98 if team_id: 

99 search_scope += f" (team: {team_id})" 

100 if project_id: 

101 search_scope += f" (project: {project_id})" 

102 return search_scope 

103 

104 

105def _format_basic_stats(stats: dict[str, Any]) -> str: 

106 """Format basic team statistics.""" 

107 return ( 

108 f"**Members**: {stats.get('member_count', 0)}\n" 

109 f"**Reflections**: {stats.get('reflection_count', 0)}\n" 

110 f"**Projects**: {stats.get('project_count', 0)}\n" 

111 f"**Total Votes**: {stats.get('total_votes', 0)}\n\n" 

112 ) 

113 

114 

115def _format_activity_stats(stats: dict[str, Any]) -> str: 

116 """Format recent activity statistics.""" 

117 if not stats.get("recent_activity"): 

118 return "" 

119 

120 output = "**Recent Activity**:\n" 

121 for activity in stats["recent_activity"][:5]: 

122 output += ( 

123 f"- {activity.get('timestamp', '')}: {activity.get('description', '')}\n" 

124 ) 

125 return output 

126 

127 

128def _format_contributor_stats(stats: dict[str, Any]) -> str: 

129 """Format top contributors statistics.""" 

130 if not stats.get("top_contributors"): 

131 return "" 

132 

133 output = "\n**Top Contributors**:\n" 

134 for contributor in stats["top_contributors"][:5]: 

135 username = contributor.get("username", "") 

136 contributions = contributor.get("contributions", 0) 

137 output += f"- {username}: {contributions} contributions\n" 

138 return output 

139 

140 

141def _format_popular_tags(stats: dict[str, Any]) -> str: 

142 """Format popular tags statistics.""" 

143 if not stats.get("popular_tags"): 

144 return "" 

145 

146 tags = ", ".join(stats["popular_tags"][:10]) 

147 return f"\n**Popular Tags**: {tags}\n" 

148 

149 

150def _format_team_statistics(team_id: str, stats: dict[str, Any]) -> str: 

151 """Format team statistics for display.""" 

152 output = f"📊 **Team Statistics: {team_id}**\n\n" 

153 

154 output += _format_basic_stats(stats) 

155 output += _format_activity_stats(stats) 

156 output += _format_contributor_stats(stats) 

157 output += _format_popular_tags(stats) 

158 

159 return output 

160 

161 

162# ============================================================================ 

163# Team Operation Implementations 

164# ============================================================================ 

165 

166 

167async def _create_team_operation( 

168 manager: Any, 

169 team_id: str, 

170 name: str, 

171 description: str, 

172 owner_id: str, 

173) -> str: 

174 """Create a new team for knowledge sharing.""" 

175 await manager.create_team( 

176 team_id=team_id, 

177 name=name, 

178 description=description, 

179 owner_id=owner_id, 

180 ) 

181 return f"✅ Team created successfully: {name}" 

182 

183 

184async def _create_team_impl( 

185 team_id: str, 

186 name: str, 

187 description: str, 

188 owner_id: str, 

189) -> str: 

190 """Create a new team for knowledge sharing.""" 

191 

192 async def operation(manager: Any) -> str: 

193 return await _create_team_operation( 

194 manager, team_id, name, description, owner_id 

195 ) 

196 

197 return await _execute_team_operation( 

198 "Create team", 

199 operation, 

200 ) 

201 

202 

203async def _search_team_knowledge_operation( 

204 manager: Any, 

205 query: str, 

206 user_id: str, 

207 team_id: str | None, 

208 project_id: str | None, 

209 tags: list[str] | None, 

210 limit: int, 

211) -> str: 

212 """Search team reflections with access control.""" 

213 results = await manager.search_team_reflections( 

214 query=query, 

215 user_id=user_id, 

216 team_id=team_id, 

217 project_id=project_id, 

218 tags=tags, 

219 limit=limit, 

220 ) 

221 

222 if not results: 

223 search_scope = _format_search_scope(team_id, project_id) 

224 return f"🔍 No results found in {search_scope} for: {query}" 

225 

226 output = f"🔍 **{len(results)} team knowledge results** for '{query}'\n\n" 

227 

228 for i, result in enumerate(results, 1): 

229 output += _format_search_result(result, i) 

230 

231 return output 

232 

233 

234async def _search_team_knowledge_impl( 

235 query: str, 

236 user_id: str, 

237 team_id: str | None = None, 

238 project_id: str | None = None, 

239 tags: list[str] | None = None, 

240 limit: int = 20, 

241) -> str: 

242 """Search team reflections with access control.""" 

243 

244 async def operation(manager: Any) -> str: 

245 return await _search_team_knowledge_operation( 

246 manager, 

247 query, 

248 user_id, 

249 team_id, 

250 project_id, 

251 tags, 

252 limit, 

253 ) 

254 

255 return await _execute_team_operation( 

256 "Search team knowledge", 

257 operation, 

258 ) 

259 

260 

261async def _get_team_statistics_operation( 

262 manager: Any, 

263 team_id: str, 

264 user_id: str, 

265) -> str: 

266 """Get team statistics and activity.""" 

267 stats = await manager.get_team_stats(team_id=team_id, user_id=user_id) 

268 

269 if not stats: 

270 return "❌ Failed to retrieve team statistics" 

271 

272 return _format_team_statistics(team_id, stats) 

273 

274 

275async def _get_team_statistics_impl(team_id: str, user_id: str) -> str: 

276 """Get team statistics and activity.""" 

277 

278 async def operation(manager: Any) -> str: 

279 return await _get_team_statistics_operation(manager, team_id, user_id) 

280 

281 return await _execute_team_operation( 

282 "Get team statistics", 

283 operation, 

284 ) 

285 

286 

287async def _vote_on_reflection_operation( 

288 manager: Any, 

289 reflection_id: str, 

290 user_id: str, 

291 vote_delta: int, 

292) -> str: 

293 """Vote on a team reflection (upvote/downvote).""" 

294 result = await manager.vote_reflection( 

295 reflection_id=reflection_id, 

296 user_id=user_id, 

297 vote_delta=vote_delta, 

298 ) 

299 

300 if result: 

301 return "✅ Reflection voted on successfully\n📊 Vote recorded\n" 

302 return "❌ Failed to vote on reflection" 

303 

304 

305async def _vote_on_reflection_impl( 

306 reflection_id: str, 

307 user_id: str, 

308 vote_delta: int = 1, 

309) -> str: 

310 """Vote on a team reflection (upvote/downvote).""" 

311 

312 async def operation(manager: Any) -> str: 

313 return await _vote_on_reflection_operation( 

314 manager, reflection_id, user_id, vote_delta 

315 ) 

316 

317 return await _execute_team_operation( 

318 "Vote on reflection", 

319 operation, 

320 ) 

321 

322 

323# ============================================================================ 

324# MCP Tool Registration 

325# ============================================================================ 

326 

327 

328def register_team_tools(mcp: FastMCP) -> None: 

329 """Register all team collaboration MCP tools. 

330 

331 Args: 

332 mcp: FastMCP server instance 

333 

334 """ 

335 

336 @mcp.tool() 

337 async def create_team( 

338 team_id: str, 

339 name: str, 

340 description: str, 

341 owner_id: str, 

342 ) -> str: 

343 """Create a new team for knowledge sharing.""" 

344 return await _create_team_impl(team_id, name, description, owner_id) 

345 

346 @mcp.tool() 

347 async def search_team_knowledge( 

348 query: str, 

349 user_id: str, 

350 team_id: str | None = None, 

351 project_id: str | None = None, 

352 tags: list[str] | None = None, 

353 limit: int = 20, 

354 ) -> str: 

355 """Search team reflections with access control.""" 

356 return await _search_team_knowledge_impl( 

357 query, 

358 user_id, 

359 team_id, 

360 project_id, 

361 tags, 

362 limit, 

363 ) 

364 

365 @mcp.tool() 

366 async def get_team_statistics(team_id: str, user_id: str) -> str: 

367 """Get team statistics and activity.""" 

368 return await _get_team_statistics_impl(team_id, user_id) 

369 

370 @mcp.tool() 

371 async def vote_on_reflection( 

372 reflection_id: str, 

373 user_id: str, 

374 vote_delta: int = 1, 

375 ) -> str: 

376 """Vote on a team reflection (upvote/downvote).""" 

377 return await _vote_on_reflection_impl(reflection_id, user_id, vote_delta)