mitmproxy-mcp MCP Server

Instructions

Implementation Reference

• src/mitmproxy_mcp/server.py:163-261 (handler)

  The core handler function for the 'get_flow_details' tool. It retrieves specific flows by index from a mitmproxy session dump, parses JSON content with size-based truncation or structure previews for large payloads, compiles detailed information including headers and content, and returns a JSON-formatted list of flow details.
  async def get_flow_details(arguments: dict) -> list[types.TextContent]: """ Gets details of specific flows from a mitmproxy dump file. For large JSON content, returns structure preview instead of full content. """ session_id = arguments.get("session_id") flow_indexes = arguments.get("flow_indexes") include_content = arguments.get("include_content", True) if not session_id: return [types.TextContent(type="text", text="Error: Missing session_id")] if not flow_indexes: return [types.TextContent(type="text", text="Error: Missing flow_indexes")] try: flows = await get_flows_from_dump(session_id) flow_details_list = [] for flow_index in flow_indexes: try: flow = flows[flow_index] if flow.type == "http": request = flow.request response = flow.response # Parse content request_content = parse_json_content(request.content, dict(request.headers)) response_content = None if response: response_content = parse_json_content(response.content, dict(response.headers)) # Handle large content request_content_preview = None response_content_preview = None flow_details = {} # Check if request content is large and is JSON if include_content and len(request.content) > MAX_CONTENT_SIZE and isinstance(request_content, dict): request_content_preview = generate_json_structure(request_content) request_content = None # Don't include full content elif include_content and len(request.content) > MAX_CONTENT_SIZE: if isinstance(request_content, str): request_content = request_content[:MAX_CONTENT_SIZE] + " ...[truncated]" else: request_content = request_content[:MAX_CONTENT_SIZE].decode(errors="ignore") + " ...[truncated]" flow_details["request_content_note"] = f"Request content truncated to {MAX_CONTENT_SIZE} bytes." # Check if response content is large and is JSON if response and include_content and len(response.content) > MAX_CONTENT_SIZE and isinstance(response_content, dict): response_content_preview = generate_json_structure(response_content) response_content = None # Don't include full content elif response and include_content and len(response.content) > MAX_CONTENT_SIZE: if isinstance(response_content, str): response_content = response_content[:MAX_CONTENT_SIZE] + " ...[truncated]" else: response_content = response_content[:MAX_CONTENT_SIZE].decode(errors="ignore") + " ...[truncated]" flow_details["response_content_note"] = f"Response content truncated to {MAX_CONTENT_SIZE} bytes." # Build flow details flow_details.update( { "index": flow_index, "method": request.method, "url": request.url, "request_headers": dict(request.headers), "status": response.status_code if response else None, "response_headers": dict(response.headers) if response else None, }) # Add content or previews based on size if include_content: if request_content is not None: flow_details["request_content"] = request_content if request_content_preview is not None: flow_details["request_content_preview"] = request_content_preview flow_details["request_content_size"] = len(request.content) flow_details["request_content_note"] = "Content too large to display. Use extract_json_fields tool to get specific values." if response_content is not None: flow_details["response_content"] = response_content if response_content_preview is not None: flow_details["response_content_preview"] = response_content_preview flow_details["response_content_size"] = len(response.content) if response else 0 flow_details["response_content_note"] = "Content too large to display. Use extract_json_fields tool to get specific values." flow_details_list.append(flow_details) else: flow_details_list.append({"error": f"Flow {flow_index} is not an HTTP flow"}) except IndexError: flow_details_list.append({"error": f"Flow index {flow_index} out of range"}) return [types.TextContent(type="text", text=json.dumps(flow_details_list, indent=2))] except FileNotFoundError: return [types.TextContent(type="text", text="Error: Session not found")] except Exception as e: return [types.TextContent(type="text", text=f"Error reading flow details: {str(e)}")]
• src/mitmproxy_mcp/server.py:52-77 (registration)

  Registration of the 'get_flow_details' tool within the list_tools() function, including name, description, and input JSON Schema defining required session_id and flow_indexes parameters.
  types.Tool( name="get_flow_details", description="Lists HTTP requests/responses from a mitmproxy capture session, showing method, URL, and status codes", inputSchema={ "type": "object", "properties": { "session_id": { "type": "string", "description": "The ID of the session" }, "flow_indexes": { "type": "array", "items": { "type": "integer" }, "description": "The indexes of the flows" }, "include_content": { "type": "boolean", "description": "Whether to include full content in the response (default: true)", "default": True } }, "required": ["session_id", "flow_indexes"] } ),
• src/mitmproxy_mcp/server.py:55-75 (schema)

  JSON Schema for 'get_flow_details' tool inputs, specifying object with required string session_id, array of integer flow_indexes, and optional boolean include_content.
  inputSchema={ "type": "object", "properties": { "session_id": { "type": "string", "description": "The ID of the session" }, "flow_indexes": { "type": "array", "items": { "type": "integer" }, "description": "The indexes of the flows" }, "include_content": { "type": "boolean", "description": "Whether to include full content in the response (default: true)", "default": True } }, "required": ["session_id", "flow_indexes"]
• src/mitmproxy_mcp/server.py:412-413 (helper)

  Dispatch logic in the generic @server.call_tool() handler that routes 'get_flow_details' calls to the specific handler function.
  elif name == "get_flow_details": return await get_flow_details(arguments)