mcpcap

Instructions

Implementation Reference

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

  The handler function for the 'analyze_tcp_connections' tool. It defines the input parameters, documentation, and delegates to the analyze_packets method with analysis_type='connections' for core processing.
  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)

  The registration block where the analyze_tcp_connections tool (and related TCP tools) are registered with the MCP server using self.mcp.tool().
  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 that performs the detailed TCP connection analysis: groups packets by connection, analyzes each connection's state, handshakes, resets, retransmissions, and compiles summary 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, }
• src/mcpcap/modules/tcp.py:63-163 (helper)

  Dispatcher helper method that sets the analysis type and kwargs, then calls the base class's analyze_packets method, which leads to _analyze_protocol_file routing to specific analyzers.
  ) def analyze_tcp_anomalies( self, pcap_file: str, server_ip: Optional[str] = None, server_port: Optional[int] = None, ) -> dict[str, Any]: """ Detect TCP traffic patterns through statistical analysis. This tool analyzes TCP traffic to identify observable patterns without making assumptions about root causes. It provides factual metrics and pattern detection that can be used for further investigation. 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 Returns: A structured dictionary containing: - statistics: Comprehensive TCP metrics (handshakes, flags, RST distribution, etc.) - patterns: Observable patterns detected in the traffic - summary: High-level summary of findings Detected pattern categories: - connection_establishment: Handshake success/failure rates, SYN response ratios - connection_termination: RST distribution, normal vs abnormal closes - reliability: Retransmission rates, packet loss indicators - connection_lifecycle: Connection state transitions The analysis is purely observational - it reports what is seen in the traffic without attempting to diagnose specific issues like "firewall block" or "network congestion". This allows the data to be interpreted in context. """ return self.analyze_packets( pcap_file, analysis_type="anomalies", server_ip=server_ip, server_port=server_port, ) def analyze_tcp_retransmissions( self, pcap_file: str, server_ip: Optional[str] = None, threshold: float = 0.02, ) -> dict[str, Any]: """ Analyze TCP retransmission patterns. Args: pcap_file: HTTP URL or absolute local file path to PCAP file server_ip: Optional filter for server IP address threshold: Retransmission rate threshold (default: 2%) Returns: A structured dictionary containing: - total_retransmissions: Total number of retransmissions - retransmission_rate: Overall retransmission rate - by_connection: Per-connection retransmission statistics - summary: Worst connections and threshold violations """ return self.analyze_packets( pcap_file, analysis_type="retransmissions", server_ip=server_ip, threshold=threshold, ) def analyze_traffic_flow( self, pcap_file: str, server_ip: str, server_port: Optional[int] = None, ) -> dict[str, Any]: """ Analyze bidirectional traffic flow characteristics. Identifies traffic direction, asymmetry, RST sources, and data transfer patterns. Args: pcap_file: HTTP URL or absolute local file path to PCAP file server_ip: Server IP address (required) server_port: Optional filter for server port Returns: A structured dictionary containing: - client_to_server: Client-to-server traffic statistics - server_to_client: Server-to-client traffic statistics - analysis: Asymmetry analysis and interpretations """ return self.analyze_packets( pcap_file, analysis_type="traffic_flow", server_ip=server_ip, server_port=server_port, ) def analyze_packets(