System Information MCP Server

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