kubecon-eu-mcp

Instructions

Implementation Reference

• src/kubecon_eu_mcp/server.py:383-416 (handler)

  The MCP tool handler for 'detect_conflicts' which parses inputs and calls the data_service logic.

  @mcp.tool()
  async def detect_conflicts(session_uids: str) -> str:
      """Detect scheduling conflicts among selected sessions.
  
      Checks whether any of the provided sessions overlap in time, helping
      attendees resolve conflicts in their planned schedule.
  
      Inspired by kubecon-event-scorer's conflict detection.
  
      Args:
          session_uids: Comma-separated session UIDs to check for conflicts.
  
      Returns:
          JSON with conflict pairs, overlap duration, and session details.
      """
      uids = [uid.strip() for uid in session_uids.split(",") if uid.strip()]
      if len(uids) < 2:
          return json.dumps(
              {"message": "Need at least 2 session UIDs to check for conflicts."}
          )
  
      conflicts = await data_service.detect_conflicts(uids)
  
      return json.dumps(
          {
              "sessions_checked": len(uids),
              "conflicts_found": len(conflicts),
              "conflicts": conflicts,
              "tip": "Use `search_sessions` to find alternative sessions on the same topic."
              if conflicts
              else "No conflicts — your schedule is clear!",
          },
          indent=2,
      )

• src/kubecon_eu_mcp/data_service.py:253-308 (handler)

  The actual business logic implementation for 'detect_conflicts' inside the DataService class.

  async def detect_conflicts(self, session_uids: list[str]) -> list[dict]:
      """Detect scheduling conflicts among selected sessions.
  
      Inspired by kubecon-event-scorer's conflicts_with() method.
  
      Args:
          session_uids: List of session UIDs to check for conflicts.
  
      Returns:
          List of conflict pairs with details.
      """
      all_sessions = await self.get_sessions()
      colocated = await self.get_colocated_sessions()
      session_map = {s.uid: s for s in all_sessions + colocated}
  
      selected = [session_map[uid] for uid in session_uids if uid in session_map]
      conflicts: list[dict] = []
  
      for i, a in enumerate(selected):
          for b in selected[i + 1 :]:
              if a.day != b.day:
                  continue
              # Parse ISO times and check overlap
              try:
                  a_start = datetime.fromisoformat(a.start)
                  a_end = datetime.fromisoformat(a.end)
                  b_start = datetime.fromisoformat(b.start)
                  b_end = datetime.fromisoformat(b.end)
              except (ValueError, TypeError):
                  continue
  
              if a_start < b_end and b_start < a_end:
                  overlap_start = max(a_start, b_start)
                  overlap_end = min(a_end, b_end)
                  overlap_min = int(
                      (overlap_end - overlap_start).total_seconds() / 60
                  )
                  conflicts.append(
                      {
                          "session_a": {
                              "uid": a.uid,
                              "title": a.title,
                              "time": f"{a.start} - {a.end}",
                              "location": a.location,
                          },
                          "session_b": {
                              "uid": b.uid,
                              "title": b.title,
                              "time": f"{b.start} - {b.end}",
                              "location": b.location,
                          },
                          "overlap_minutes": overlap_min,
                      }
                  )
  
      return conflicts