mcpacket

Instructions

Implementation Reference

• src/mcpcap/modules/tcp.py:21-63 (handler)

  The primary handler function for the 'analyze_tcp_connections' tool. Defines input parameters, documentation serving as schema, and delegates to analyze_packets with 'connections' analysis type for execution.
  def analyze_tcp_connections( self, pcap_file: str, server_ip: Optional[str] = None, server_port: Optional[int] = None, detailed: bool = False, ) -> dict[str, Any]: """ Analyze TCP connection states and lifecycle. This is the core tool for TCP connection analysis, solving 80% of TCP-related issues. ⚠️ FILE UPLOAD LIMITATION: This MCP tool cannot process files uploaded through Claude's web interface. Files must be accessible via URL or local file path. SUPPORTED INPUT FORMATS: - Remote files: "https://example.com/capture.pcap" - Local files: "/absolute/path/to/capture.pcap" UNSUPPORTED: - Files uploaded through Claude's file upload feature - Base64 file content - Relative file paths Args: pcap_file: HTTP URL or absolute local file path to PCAP file server_ip: Optional filter for server IP address server_port: Optional filter for server port detailed: Whether to return detailed connection information Returns: A structured dictionary containing TCP connection analysis results including: - summary: Overall connection statistics - connections: List of individual connections with states - issues: Detected problems """ return self.analyze_packets( pcap_file, analysis_type="connections", server_ip=server_ip, server_port=server_port, detailed=detailed, )
• src/mcpcap/core/server.py:58-62 (registration)

  Registration of the 'analyze_tcp_connections' tool (alongside other TCP analysis tools) to the FastMCP server instance.
  elif module_name == "tcp": self.mcp.tool(module.analyze_tcp_connections) self.mcp.tool(module.analyze_tcp_anomalies) self.mcp.tool(module.analyze_tcp_retransmissions) self.mcp.tool(module.analyze_traffic_flow)
• src/mcpcap/modules/tcp.py:234-301 (helper)

  Core helper function implementing the TCP connection analysis logic, grouping packets by 4-tuple, analyzing each connection's state, handshakes, resets, and compiling summary statistics and issues.
  def _analyze_connections( self, pcap_file: str, tcp_packets: list, all_packets: list ) -> dict[str, Any]: """Analyze TCP connections.""" # Group packets by connection (4-tuple) connections = defaultdict(list) for pkt in tcp_packets: conn_key = self._get_connection_key(pkt) connections[conn_key].append(pkt) # Analyze each connection connection_details = [] successful_handshakes = 0 failed_handshakes = 0 reset_connections = 0 normal_close = 0 issues = [] for conn_key, pkts in connections.items(): conn_info = self._analyze_single_connection(conn_key, pkts) connection_details.append(conn_info) if conn_info["handshake_completed"]: successful_handshakes += 1 else: failed_handshakes += 1 if conn_info["close_reason"] == "reset": reset_connections += 1 elif conn_info["close_reason"] == "normal": normal_close += 1 # Detect issues total_rst = sum(c["rst_count"] for c in connection_details) total_retrans = sum(c["retransmissions"] for c in connection_details) if reset_connections > 0: issues.append(f"{reset_connections} connections terminated by RST") if total_retrans > 0: issues.append(f"{total_retrans} retransmissions detected") if failed_handshakes > 0: issues.append(f"{failed_handshakes} failed handshakes") return { "file": pcap_file, "analysis_timestamp": datetime.now().isoformat(), "total_packets": len(all_packets), "tcp_packets_found": len(tcp_packets), "filter": { "server_ip": self._analysis_kwargs.get("server_ip"), "server_port": self._analysis_kwargs.get("server_port"), }, "summary": { "total_connections": len(connections), "successful_handshakes": successful_handshakes, "failed_handshakes": failed_handshakes, "established_connections": successful_handshakes, "reset_connections": reset_connections, "normal_close": normal_close, "active_connections": len(connections) - reset_connections - normal_close, }, "connections": connection_details if self._analysis_kwargs.get("detailed", False) else connection_details[:10], "issues": issues, }