get_open_ports

Instructions

Implementation Reference

• src/sysinfo/server.py:192-208 (handler)

  The handler function for the "get_open_ports" tool, registered via @mcp.tool decorator. It wraps the get_network_ports() helper, adds headers/timestamps/error handling, and returns a formatted ToolResult.

  @mcp.tool
  def get_open_ports() -> ToolResult:
      """Get list of open network ports and listening services.
      
      Shows listening ports with associated processes and active network connections.
      Critical for security analysis, service monitoring, and network troubleshooting.
      """
      info_sections = []
      info_sections.append("# Open Network Ports")
      info_sections.append(f"*Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}*\n")
      
      try:
          info_sections.extend(get_network_ports())
      except Exception as e:
          info_sections.append(f"⚠️ **Port detection error**: {str(e)}")
      
      return text_response("\n".join(info_sections))

• src/sysinfo/collectors.py:885-999 (helper)

  The supporting utility function get_network_ports() that implements the core logic for detecting open/listening ports and active connections using psutil.net_connections(). Handles permissions fallbacks, formats markdown tables with process details, and provides summaries.

  def get_network_ports() -> List[str]:
      """Get list of open network ports and listening services"""
      info = []
      info.append("## 🌐 Network Ports")
      
      try:
          # Get network connections - try with different approaches for permissions
          try:
              connections = psutil.net_connections(kind='inet')
          except psutil.AccessDenied:
              # Fallback to connections without process info
              try:
                  connections = psutil.net_connections(kind='inet', pid=None)
              except Exception:
                  # Last resort - get connections for current process only
                  connections = psutil.Process().connections(kind='inet')
          
          # Group by listening ports
          listening_ports = {}
          established_connections = []
          
          for conn in connections:
              try:
                  if conn.status == psutil.CONN_LISTEN and conn.laddr:
                      # This is a listening port
                      port_key = f"{conn.laddr.ip}:{conn.laddr.port}"
                      if port_key not in listening_ports:
                          # Get process info
                          proc_name = "Unknown"
                          proc_exe = "N/A"
                          if conn.pid:
                              try:
                                  proc = psutil.Process(conn.pid)
                                  proc_name = proc.name()
                                  try:
                                      proc_exe = proc.exe() or "N/A"
                                  except (psutil.AccessDenied, OSError):
                                      proc_exe = "Access denied"
                              except (psutil.NoSuchProcess, psutil.AccessDenied, OSError):
                                  proc_name = "Access denied"
                          
                          listening_ports[port_key] = {
                              'port': conn.laddr.port,
                              'ip': conn.laddr.ip,
                              'protocol': 'TCP' if conn.type == 1 else 'UDP',
                              'process': proc_name,
                              'executable': proc_exe,
                              'pid': conn.pid
                          }
                  
                  elif conn.status == psutil.CONN_ESTABLISHED and conn.laddr and conn.raddr:
                      # This is an established connection
                      established_connections.append({
                          'local': f"{conn.laddr.ip}:{conn.laddr.port}",
                          'remote': f"{conn.raddr.ip}:{conn.raddr.port}",
                          'pid': conn.pid
                      })
                      
              except Exception:
                  continue
          
          # Display listening ports
          if listening_ports:
              info.append("\n### Listening Ports")
              info.append("| Port | Protocol | Bind Address | Process | PID |")
              info.append("|------|----------|--------------|---------|-----|")
              
              # Sort by port number
              sorted_ports = sorted(listening_ports.values(), key=lambda x: x['port'])
              
              for port_info in sorted_ports:
                  port = port_info['port']
                  protocol = port_info['protocol']
                  ip = port_info['ip']
                  process = port_info['process'][:25]  # Truncate long process names
                  pid = port_info['pid'] or 'N/A'
                  
                  # Show bind address (0.0.0.0 means all interfaces)
                  bind_addr = "All interfaces" if ip == "0.0.0.0" else ip
                  if ip == "127.0.0.1":
                      bind_addr = "Localhost only"
                  elif ip.startswith("::"):
                      bind_addr = "IPv6"
                  
                  info.append(f"| {port} | {protocol} | {bind_addr} | {process} | {pid} |")
          
          # Display established connections summary (top 10)
          if established_connections:
              info.append(f"\n### Active Connections")
              info.append(f"**Total established connections**: {len(established_connections)}")
              
              # Group by remote host
              remote_hosts = {}
              for conn in established_connections:
                  remote_ip = conn['remote'].split(':')[0]
                  if remote_ip not in remote_hosts:
                      remote_hosts[remote_ip] = 0
                  remote_hosts[remote_ip] += 1
              
              # Show top remote hosts
              if remote_hosts:
                  info.append("\n**Top remote hosts by connection count**:")
                  sorted_hosts = sorted(remote_hosts.items(), key=lambda x: x[1], reverse=True)
                  for host, count in sorted_hosts[:10]:
                      info.append(f"- {host}: {count} connection{'s' if count > 1 else ''}")
          
          # Add summary
          total_listening = len(listening_ports)
          total_established = len(established_connections)
          info.append(f"\n**Summary**: {total_listening} listening ports, {total_established} active connections")
          
      except Exception as e:
          info.append(f"⚠️ **Error collecting network port information**: {str(e)}")
      
      return info